ThreadPoolExecutor 簡介

ThreadPoolExecutor 是 java.util.concurrent 包下的一個類，在jdk1.5版本引入，幫助開發人員管理執行緒並方便地執行並行任務。

通俗來說，ThreadPoolExecutor 的作用是生產和管理執行緒池的，可以通過呼叫其 execute 方法和 submit 方法執行多執行緒任務。

ThreadPoolExecutor 使用

建立執行器

ExecutorService 物件和 ThreadPoolExecutor 的關係如下圖：

ExecutorServiceConfig：

package com.ramble.threadpool.config;import java.util.concurrent.*;



public class ExecutorServiceConfig {

    /**
     * 定義一個並行任務執行器服務
     */
    private static ExecutorService executorService;
    /**
     * 在類載入的時候初始化並行任務執行器
     */
    static {
        init();
    }
    
    /**
     * 防止類屬性被篡改
     */
    private ExecutorServiceConfig() {
    }
    
    /**
     * 初始化並行任務執行器。核心執行緒數量：設定為2，初始建立的執行緒池大小；最大執行緒數量：設定為3；空閒執行緒存活時間：設定為3秒，當非核心執行緒執行完任務之後，若沒有新的任務分派，存活多久後自動銷燬；任務佇列：設定為2，當執行緒池建立的執行緒數量達到最大執行緒數量後，新進來的任務會排隊等候；
     * 拒絕策略：設定為直接拋異常
     * <p>
     * 以上設定需要根據：實際的業務場景、專案實際情況、實際硬體情況等各種因素綜合考量
     */
    private static void init() {
        executorService = new ThreadPoolExecutor(2, 3, 3, TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>(2), new ThreadPoolExecutor.AbortPolicy());
    }
    
    /**
     * 獲取預設並行任務執行器
     *
     * @return
     */
    public static ExecutorService getDefaultExecutor() {
        return executorService;
    }
    
    /**
     * 獲取固定大小並行任務執行器
     *
     * @return
     */
    public static ExecutorService getFixedExecutor() {
        return Executors.newFixedThreadPool(10);
    }
    
    /**
     * 獲取其他並行任務執行器
     * @return
     */
    public static ExecutorService getOtherExecutor() {
        //todo
        return null;
    }
}

這個類的核心目的是構造一個 ExecutorService ，供業務程式碼呼叫。當業務程式碼需要建立執行緒執行任務的時候，不用建立、管理和維護「執行者（執行緒）」，只需要告訴 ExecutorService 需要做什麼「事情（任務）」。
ExecutorService 物件必須是單例的，因為此物件本身是管理執行緒池的，如果自己都不是執行緒安全的，那使用起來將有可能發生災難。
一個java程序允許建立多個ExecutorService 。根據業務實際情況，如果業務邏輯比較單一，大概率建立一種就滿足使用。若業務繁雜，可根據業務特性抽象出多種型別以滿足不同需求。譬如程式碼中就建立了兩個 getDefaultExecutor 和 getFixedExecutor
關於執行緒池設定，如核心執行緒數量、最大執行緒數量、拒絕策略等等沒有絕對正確的值做參考，需要根據實際情況設定

建立任務

通過實現 Runnable 介面並重寫 run 方法建立無返回值的多執行緒任務
通過實現 Callable 介面並重寫 call 方法建立有返回值的多執行緒任務

Task1:


package com.ramble.threadpool.task;
import lombok.extern.slf4j.Slf4j;
@Slf4j
public class Task1 implements Runnable {
    /**
     * 新執行緒執行一個任務，沒有引數，不需要返回值，此任務和其他任務沒有先後順序
     */
    @Override
    public void run() {
        try {
            Thread.sleep(3000L);
            log.info("TaskOne,thread is ={},thread name is={}", Thread.currentThread().getId(), Thread.currentThread().getName());
            throw new IllegalAccessException("主動拋一個異常");
        } catch (IllegalAccessException e) {
            throw new RuntimeException(e);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}

Task2：


package com.ramble.threadpool.task;
import com.alibaba.fastjson.JSON;
import com.ramble.threadpool.dto.TaskTwoParam;
import lombok.extern.slf4j.Slf4j;
@Slf4j
public class Task2 implements Runnable {
    private TaskTwoParam param;
    public Task2(TaskTwoParam param) {
        this.param = param;
    }
    /**
     * 新執行緒執行一個任務，引數通過建構函式傳遞，不需要返回值，此任務和其他任務沒有先後順序
     */
    @Override
    public void run() {
        try {
            Thread.sleep(3000L);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        log.info("TaskTwo,thread  id is ={},thread name is={}, param is ={}", Thread.currentThread().getId(), Thread.currentThread().getName(), JSON.toJSONString(param));
    }
}

Task3：


package com.ramble.threadpool.task;
import com.alibaba.fastjson.JSON;
import com.ramble.threadpool.dto.TaskDto;
import com.ramble.threadpool.dto.TaskTwoParam;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.Callable;
@Slf4j
public class Task3 implements Callable {
    private TaskTwoParam param;
    public Task3(TaskTwoParam param) {
        this.param = param;
    }
    /**
     * 新執行緒執行一個任務，通過建構函式傳遞引數，有返回值，此任務和其他任務沒有先後順序
     *
     * @return
     * @throws Exception
     */
    @Override
    public Object call() throws Exception {
        log.info("Task3,thread  id is ={},thread name is={}, param is ={}", Thread.currentThread().getId(), Thread.currentThread().getName(), JSON.toJSONString(param));
        return new TaskDto().setName("task3-callable");
    }
}

Task4：


package com.ramble.threadpool.task;
import lombok.extern.slf4j.Slf4j;
@Slf4j
public class Task4 implements Runnable {
    /**
     * 新執行緒執行一個任務，沒有引數，不需要返回值，此任務和其他任務沒有先後順序
     */
    @Override
    public void run() {
        try {
            Thread.sleep(3000L);
            log.info("Task4,thread is ={},thread name is={}", Thread.currentThread().getId(), Thread.currentThread().getName());
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}

執行任務

在 controller 中模擬發起多執行緒任務。


package com.ramble.threadpool.controller;
import com.alibaba.fastjson.JSON;
import com.ramble.threadpool.config.ExecutorServiceConfig;
import com.ramble.threadpool.dto.TaskDto;
import com.ramble.threadpool.dto.TaskTwoParam;
import com.ramble.threadpool.task.Task1;
import com.ramble.threadpool.task.Task2;
import com.ramble.threadpool.task.Task3;
import com.ramble.threadpool.task.Task4;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.task.TaskExecutionProperties;
import org.springframework.scheduling.annotation.Async;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
   
   
   
    @GetMapping("/task")
    public String testTask() {
        log.info("testTask,thread is ={},thread name is={}", Thread.currentThread().getId(), Thread.currentThread().getName());
//        for (int i = 0; i < 4; i++) {
//            ExecutorServiceConfig.getExecutor().execute(new Task1());
//        }
//
//        for (int i = 0; i < 5; i++) {
//            ExecutorServiceConfig.getExecutor().execute(new Task2(new TaskTwoParam().setId(100).setName("cnaylor")));
//        }
        ExecutorServiceConfig.getDefaultExecutor().execute(new Task1());
        ExecutorServiceConfig.getDefaultExecutor().execute(new Task2(new TaskTwoParam().setId(100).setName("cnaylor")));
        ExecutorServiceConfig.getFixedExecutor().execute(new Task4());
        Future<TaskDto> taskResult = ExecutorServiceConfig.getDefaultExecutor().submit(new Task3(new TaskTwoParam().setId(100).setName("cnaylor")));
        TaskDto taskDto;
        try {
            taskDto = taskResult.get();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (ExecutionException e) {
            throw new RuntimeException(e);
        }
        return "testTask";
    }

ThreadPoolExecutor 拒絕策略

所謂拒絕策略，一般是指當前執行緒池處於滿負荷狀態，所有的執行緒都有正在處理的任務，阻塞佇列也排滿了的情況下，對新進來的任務做出何種響應。

預設的拒絕策略有四種，同時也可以自定義拒絕策略。

AbortPolicy
CallerRunsPolicy
DiscardOldestPolicy
DiscardPolicy

AbortPolicy

拋一個異常（RejectedExecutionException），此任務將無法執行。

CallerRunsPolicy

執行緒池滿載了，此任務將由呼叫發起執行緒來執行。比如我們在一個http請求執行緒中呼叫了執行緒池處理非同步任務，但是現線上程池滿了，那麼此任務將轉由http請求執行緒處理。缺點是會導致http請求執行緒阻塞，達不到非同步處理的效果。優點是任務會正常執行，避免被任務執行器丟棄。

DiscardOldestPolicy

在阻塞佇列最前端拋棄一個任務，然後將此任務新增到阻塞佇列中排隊。最前端是指最先新增到阻塞佇列的任務。

DiscardPolicy

當前任務不會執行，也不會拋異常，好像什麼也沒有發生一樣。

自定義拒絕策略

自己編寫一個類，實現 RejectedExecutionHandler 介面，並重寫 rejectedExecution 方法即可實現自定義拒絕策略。需要在範例化 ThreadPoolExecutor 的時候，將自定義策略設定進去。

Java執行緒池ThreadPoolExecutor極簡教學