快捷導(dǎo)航

Java多線程并發(fā)之線程池任務(wù)請求攔截測試實(shí)例

更新時(shí)間：2023年12月14日 09:10:56 作者：Terisadeng

這篇文章主要介紹了Java多線程并發(fā)之線程池任務(wù)請求攔截測試實(shí)例,隊(duì)列中永遠(yuǎn)沒有線程被加入,即使線程池已滿,也不會導(dǎo)致被加入排隊(duì)隊(duì)列,實(shí)現(xiàn)了只有線程池存在空閑線程的時(shí)候才會接受新任務(wù)的需求,需要的朋友可以參考下

一、需求

前端會傳入一個(gè)存儲編碼的list，后臺接收到編碼通過計(jì)算返回每個(gè)編碼對應(yīng)的值，每個(gè)編碼計(jì)算出來的值是固定不變的。

二、設(shè)計(jì)方案

因?yàn)榍岸苏埱箜憫?yīng)有一個(gè)時(shí)常要求，比如100ms。

而這個(gè)計(jì)算比較耗時(shí)，因此為了請求能夠快速響應(yīng)，在第一個(gè)請求過來時(shí)判斷redis緩存是否存儲編碼對應(yīng)的計(jì)算值，如果沒有就直接返回空，前端根據(jù)這個(gè)空值使用補(bǔ)償方案的默認(rèn)值。

后臺通過線程池執(zhí)行計(jì)算方法，然后存入redis，這樣下次用戶帶著相同的編碼請求就可以直接從緩存獲取，不用重復(fù)計(jì)算。

這里的問題在于，當(dāng)并發(fā)量高的情況下，比如50個(gè)用戶帶著相同的編碼調(diào)用計(jì)算方法，而實(shí)際上計(jì)算方法只需要調(diào)用一次就可以了。

因此我們需要在將任務(wù)提交到線程池之前判斷線程池中執(zhí)行線程的數(shù)量來決定是否要將任務(wù)提交到線程池。

另外這里千萬不能使用直接創(chuàng)建線程的方式，這會導(dǎo)致并發(fā)情況下突然創(chuàng)建大量線程，導(dǎo)致系統(tǒng)cpu飆升卡死。

三、測試

1、線程池實(shí)現(xiàn)類，提供全局唯一的線程池實(shí)例

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
 * 固定大小的線程池
 */
public class DisCardThreadPool
{
    private static DisCardThreadPool disCardThreadPool=new DisCardThreadPool();
    /*
     * 將構(gòu)造方法訪問修飾符設(shè)為私有，禁止任意實(shí)例化。
     */
    private DisCardThreadPool() {
    }
    /**
     * 核心線程數(shù)
     */
    int corePoolSize = 1;
    /**
     * 最大線程數(shù)
     */
    int maximumPoolSize = 1;
    /**
     * 空閑線程存活時(shí)間
     */
    long keepAliveTime = 10;
    /*
     * 線程池單例創(chuàng)建方法
     */
    public static DisCardThreadPool newInstance() {
        return disCardThreadPool;
    }
    private final ThreadPoolExecutor mThreadPool=new ThreadPoolExecutor(corePoolSize,maximumPoolSize,keepAliveTime, TimeUnit.MILLISECONDS,new ArrayBlockingQueue<>(10), new ThreadPoolExecutor.DiscardPolicy());
    public void execute(Runnable r){
        mThreadPool.execute(r);
    }
    /**
     * 隊(duì)列中等待執(zhí)行的任務(wù)數(shù)目
     * @return
     */
    public synchronized int getQueue(){
        return mThreadPool.getQueue().size();
    }
    /*
     * 獲取線程池中剩余線程數(shù)目
     * 獲取的結(jié)果不準(zhǔn)確
     */
    public synchronized int getActiveCount(){
        return mThreadPool.getActiveCount();
    }

2、測試類

這里通過CountDownLatch類同時(shí)啟動(dòng)多個(gè)線程來模擬并發(fā)請求

import com.teriste.service.threadpool.DisCardThreadPool;
import org.junit.Test;
import java.io.File;
import java.io.IOException;
import java.util.concurrent.CountDownLatch;
/**
 * 模擬并發(fā)向線程池提交任務(wù)。
 * 需求：使得線程池滿之后其他請求都不執(zhí)行
 */
public class MultiThreadConcurrencyTest {
    //獲取線程池實(shí)例
    private static DisCardThreadPool threadPool=DisCardThreadPool.newInstance();
    @Test
    public void test(){
        //創(chuàng)建大小20的計(jì)數(shù)器，使得20個(gè)線程同時(shí)執(zhí)行，模擬并發(fā)
       CountDownLatch countDownLatch=new CountDownLatch(20);
       for (int i=0;i<20;i++){
           InvokeThread thread=new InvokeThread(countDownLatch);
           System.out.println("創(chuàng)建線程:"+thread.getName());
           thread.start();
           //啟動(dòng)一個(gè)線程，計(jì)數(shù)器就減一，同時(shí)在線程的run方法中阻塞線程，等待計(jì)數(shù)器喚醒
           countDownLatch.countDown();
       }
        try {
           //阻塞主線程，防止子線程還沒執(zhí)行主線程結(jié)束導(dǎo)致子線程無法執(zhí)行
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    //線程池執(zhí)行線程
    public static void invokeThread(){
        //當(dāng)排隊(duì)隊(duì)列有線程等待時(shí)不繼續(xù)添加線程
        synchronized (MultiThreadConcurrencyTest.class){
            //System.out.println("等待隊(duì)列大小:"+threadPool.getQueue());
            //官方api指出getActiveCount()無法獲取準(zhǔn)確的存獲線程數(shù)
            //因?yàn)檫@里是根據(jù)隊(duì)列中待執(zhí)行任務(wù)數(shù)來判斷，因此如果線程池大小為1，實(shí)際上會有兩個(gè)線程被執(zhí)行，
            //一個(gè)線程是進(jìn)入線程池，還有一個(gè)線程判斷此時(shí)隊(duì)列待執(zhí)行線程數(shù)是0會進(jìn)入待執(zhí)行隊(duì)列，因此最終執(zhí)行線程數(shù)是線程池大小+1
            System.out.println("排隊(duì)隊(duì)列中的線程個(gè)數(shù):"+threadPool.getQueue());
            if (threadPool.getQueue()<=0){
                threadPool.execute(new WorkThread());
            }
        }
    }
}
//調(diào)用線程池執(zhí)行任務(wù)的類，模擬外部請求實(shí)體發(fā)起請求
class InvokeThread extends Thread{
    private CountDownLatch countDownLatch;
    public InvokeThread(CountDownLatch countDownLatch){
        this.countDownLatch=countDownLatch;
    }
    @Override
    public void run(){
        try {
            //等待計(jì)數(shù)器喚醒
            countDownLatch.await();
            //向線程池提交線程
            MultiThreadConcurrencyTest.invokeThread();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
//任務(wù)類
class WorkThread implements Runnable{
    @Override
    public void run() {
        String path="E:\\測試";
        File file=new File(path);
        if(!file.exists()){
            file.mkdirs();//創(chuàng)建目錄
        }
        String fileName=Thread.currentThread().getName()+System.currentTimeMillis();
        File newFile=new File(path,fileName);
        try {
            newFile.createNewFile();
        } catch (IOException e) {
            e.printStackTrace();
        }
        //在線程池中有可能是不同的線程使用相同的名稱
        //因?yàn)榫€程池中上個(gè)結(jié)束的線程繼續(xù)使用來執(zhí)行下個(gè)線程
        System.out.println("當(dāng)前執(zhí)行的線程的名稱:"+Thread.currentThread().getName()+fileName);
    }
}

3、如果放開對排隊(duì)隊(duì)列的判斷可以看到，當(dāng)線程池滿了之后執(zhí)行的是ArrayBlockingQueue.offer(E e);方法：

這說明我們可以通過繼承ArrayBlockingQueue類實(shí)現(xiàn)自己的排隊(duì)隊(duì)列，當(dāng)線程池滿了之后調(diào)用offer方法時(shí)，我們直接丟棄任務(wù)什么都不做，這樣就可以準(zhǔn)確實(shí)現(xiàn)上面的方案，并且可以去掉對隊(duì)列中待執(zhí)行線程的判斷，從而不需要加鎖，提高執(zhí)行效率。

下面是自定義隊(duì)列的實(shí)現(xiàn)：

import java.util.Collection;
import java.util.concurrent.ArrayBlockingQueue;
/**
 * 線程池使用該類時(shí)執(zhí)行插入方法時(shí)不會向隊(duì)列中插入數(shù)據(jù)，會直接丟棄或記錄日志
 */
public class EmptyArrayBlockingQueue<E> extends ArrayBlockingQueue{
    public EmptyArrayBlockingQueue(int capacity) {
        super(capacity);
    }
    public EmptyArrayBlockingQueue(int capacity, boolean fair) {
        super(capacity, fair);
    }
    public EmptyArrayBlockingQueue(int capacity, boolean fair, Collection c) {
        super(capacity, fair, c);
    }
    /**
     * 注意這里重寫的父類方法參數(shù)是泛型參數(shù)
     * 由于Java的類型擦除，在編譯時(shí)會自動(dòng)變?yōu)镺bject類型
     * 因此這里使用Object類型實(shí)際上就是重寫的父類方法
     * @param e
     * @return
     */
    @Override
    public boolean offer(Object e) {
        /**不執(zhí)行將線程加入隊(duì)列的操作，這樣隊(duì)列永遠(yuǎn)為空
        超過線程池核心線程數(shù)的線程實(shí)際上在這里都被丟棄了
        可以增加記錄日志的操作
         */
        return true;
    }
}

下面是修改后的線程池類：

import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
 * 固定大小的線程池
 */
public class DisCardThreadPool
{
    private static DisCardThreadPool disCardThreadPool=new DisCardThreadPool();
    /*
     * 將構(gòu)造方法訪問修飾符設(shè)為私有，禁止任意實(shí)例化。
     */
    private DisCardThreadPool() {
    }
    /**
     * 核心線程數(shù)
     */
    int corePoolSize = 1;
    /**
     * 最大線程數(shù)
     */
    int maximumPoolSize = 1;
    /**
     * 空閑線程存活時(shí)間
     */
    long keepAliveTime = 10;
    /*
     * 線程池單例創(chuàng)建方法
     */
    public static DisCardThreadPool newInstance() {
        return disCardThreadPool;
    }
    private final ThreadPoolExecutor mThreadPool=new ThreadPoolExecutor(corePoolSize,maximumPoolSize,keepAliveTime, TimeUnit.MILLISECONDS,new EmptyArrayBlockingQueue<>(10),
            new ThreadPoolExecutor.DiscardPolicy());
    public void execute(Runnable r){
        mThreadPool.execute(r);
    }
    /**
     * 隊(duì)列中等待執(zhí)行的任務(wù)數(shù)目
     * @return
     */
    public synchronized int getQueue(){
        return mThreadPool.getQueue().size();
    }
    /*
     * 獲取線程池中剩余線程數(shù)目
     * 獲取的結(jié)果不準(zhǔn)確
     */
    public synchronized int getActiveCount(){
        return mThreadPool.getActiveCount();
    }
}

下面是測試類：

import com.teriste.service.threadpool.DisCardThreadPool;
import org.junit.Test;
import java.io.File;
import java.io.IOException;
import java.util.concurrent.CountDownLatch;
/**
 * 模擬并發(fā)向線程池提交任務(wù)。
 * 需求：使得線程池滿之后其他請求都不執(zhí)行
 */
public class MultiThreadConcurrencyTest {
    //獲取線程池實(shí)例
    private static DisCardThreadPool threadPool=DisCardThreadPool.newInstance();
    @Test
    public void test(){
        //創(chuàng)建大小20的計(jì)數(shù)器，使得20個(gè)線程同時(shí)執(zhí)行，模擬并發(fā)
       CountDownLatch countDownLatch=new CountDownLatch(20);
       for (int i=0;i<20;i++){
           InvokeThread thread=new InvokeThread(countDownLatch);
           System.out.println("創(chuàng)建線程:"+thread.getName());
           thread.start();
           //啟動(dòng)一個(gè)線程，計(jì)數(shù)器就減一，同時(shí)在線程的run方法中阻塞線程，等待計(jì)數(shù)器喚醒
           countDownLatch.countDown();
       }
        try {
           //阻塞主線程，防止子線程還沒執(zhí)行主線程結(jié)束導(dǎo)致子線程無法執(zhí)行
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    //線程池執(zhí)行線程
    public static void invokeThread(){
        //當(dāng)排隊(duì)隊(duì)列有線程等待時(shí)不繼續(xù)添加線程
        synchronized (MultiThreadConcurrencyTest.class){
            //System.out.println("等待隊(duì)列大小:"+threadPool.getQueue());
            //官方api指出getActiveCount()無法獲取準(zhǔn)確的存獲線程數(shù)
            //因?yàn)檫@里是根據(jù)隊(duì)列中待執(zhí)行任務(wù)數(shù)來判斷，因此如果線程池大小為1，實(shí)際上會有兩個(gè)線程被執(zhí)行，
            //一個(gè)線程是進(jìn)入線程池，還有一個(gè)線程判斷此時(shí)隊(duì)列待執(zhí)行線程數(shù)是0會進(jìn)入待執(zhí)行隊(duì)列，因此最終執(zhí)行線程數(shù)是線程池大小+1
            System.out.println("排隊(duì)隊(duì)列中的線程個(gè)數(shù):"+threadPool.getQueue());
            //if (threadPool.getQueue()<=0){
                threadPool.execute(new WorkThread());
            //}
        }
    }
}
//調(diào)用線程池執(zhí)行任務(wù)的類，模擬外部請求實(shí)體發(fā)起請求
class InvokeThread extends Thread{
    private CountDownLatch countDownLatch;
    public InvokeThread(CountDownLatch countDownLatch){
        this.countDownLatch=countDownLatch;
    }
    @Override
    public void run(){
        try {
            //等待計(jì)數(shù)器喚醒
            countDownLatch.await();
            //向線程池提交線程
            MultiThreadConcurrencyTest.invokeThread();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
//任務(wù)類
class WorkThread implements Runnable{
    @Override
    public void run() {
        String path="E:\\測試";
        File file=new File(path);
        if(!file.exists()){
            file.mkdirs();//創(chuàng)建目錄
        }
        String fileName=Thread.currentThread().getName()+System.currentTimeMillis();
        File newFile=new File(path,fileName);
        try {
            newFile.createNewFile();
        } catch (IOException e) {
            e.printStackTrace();
        }
        //在線程池中有可能是不同的線程使用相同的名稱
        //因?yàn)榫€程池中上個(gè)結(jié)束的線程繼續(xù)使用來執(zhí)行下個(gè)線程
        System.out.println("當(dāng)前執(zhí)行的線程的名稱:"+Thread.currentThread().getName()+fileName);
    }
}

測試結(jié)果：