ThreadPoolTaskExecutor 的使用案例

1. 依賴說明

<!-- Spring Retry（用于任務重試） -->
<dependency><groupId>org.springframework.retry</groupId><artifactId>spring-retry</artifactId><version>1.3.1</version>
</dependency>
<dependency><groupId>org.springframework</groupId><artifactId>spring-aspects</artifactId><version>5.3.22</version>
</dependency>
<!-- SLF4J MDC（用于上下文傳遞） -->
<dependency><groupId>org.slf4j</groupId><artifactId>slf4j-api</artifactId><version>1.7.36</version>
</dependency>

---

2. 完整配置代碼

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.backoff.FixedBackOff;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.backoff.FixedBackOffPolicy;
import org.springframework.core.task.TaskDecorator;import java.util.concurrent.*;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;import org.slf4j.MDC;@Configuration
public class TaskExecutorConfig {/*** 配置線程池 Bean，名稱為 "taskExecutor"* 包含上下文傳遞機制和重試機制*/@Bean(name = "taskExecutor")public ThreadPoolTaskExecutor taskExecutor() {// 創建線程池實例ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();/*** 核心線程數：線程池中始終保留的線程數* - 建議根據任務類型調整：*   - CPU密集型任務：設為 CPU 核心數（如 Runtime.getRuntime().availableProcessors()）*   - IO密集型任務：設為 2 * CPU 核心數 或更高*/executor.setCorePoolSize(5);/*** 最大線程數：線程池中允許的最大線程數* - 當隊列滿時，會創建新線程，直到達到 maxPoolSize* - 超出后觸發拒絕策略*/executor.setMaxPoolSize(10);/*** 空閑線程存活時間（單位秒）：* - 當線程數超過 corePoolSize 時，多余的線程在空閑 keepAliveSeconds 后會被回收* - 通常設置為與任務平均執行時間相當*/executor.setKeepAliveSeconds(60);/*** 任務隊列：用于緩存等待執行的任務* - ArrayBlockingQueue：有界隊列，容量為 25（可根據業務需求調整）* - LinkedBlockingQueue：無界隊列（不推薦，可能造成內存泄漏）* - SynchronousQueue：無容量隊列，適合直接傳遞任務（如 newCachedThreadPool）*/BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(25);executor.setQueue(queue);/*** 拒絕策略：當隊列滿且線程數達到最大時，新任務如何處理* - CallerRunsPolicy：由調用線程（提交任務的線程）執行任務，避免丟棄* - AbortPolicy：拋出 RejectedExecutionException（默認）* - DiscardPolicy：直接丟棄任務* - DiscardOldestPolicy：丟棄隊列中最舊的任務，嘗試重新提交當前任務*/executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());/*** 線程名稱前綴：便于日志追蹤和線程監控* - 示例輸出：TaskExecutor-1, TaskExecutor-2, ...*/executor.setThreadNamePrefix("TaskExecutor-");/*** 上下文傳遞機制：使用 TaskDecorator 傳遞 MDC 上下文（如日志追蹤 ID）* - MDC（Mapped Diagnostic Context）是 SLF4J 提供的日志上下文工具* - 通過 TaskDecorator，在任務執行前后傳遞上下文信息*/executor.setTaskDecorator(new MdcTaskDecorator());/*** 初始化線程池* - Spring 容器會自動調用此方法，但顯式調用更安全*/executor.initialize();return executor;}/*** 創建重試模板（最大重試3次，間隔1秒）* - 用于包裝任務，使其具備重試能力*/public static RetryTemplate createRetryTemplate() {RetryTemplate retryTemplate = new RetryTemplate();/*** 重試策略：定義重試條件和最大重試次數* - SimpleRetryPolicy：固定次數重試* - maxAttempts：最大重試次數（包含首次嘗試）*/SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();retryPolicy.setMaxAttempts(3); // 最多重試3次（含首次執行）retryTemplate.setRetryPolicy(retryPolicy);/*** 重試間隔策略：定義重試之間的等待時間* - FixedBackOffPolicy：固定間隔（如 1 秒）* - ExponentialBackOffPolicy：指數退避（間隔遞增）*/FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();backOffPolicy.setBackOffPeriod(1000); // 每次重試間隔 1 秒retryTemplate.setBackOffPolicy(backOffPolicy);return retryTemplate;}/*** 重試任務包裝器：將任務包裝為支持重試的CallableWrapper* - 通過 RetryTemplate 執行重試邏輯*/public static class RetryCallableWrapper implements Runnable {private final Runnable targetTask;private final RetryTemplate retryTemplate;public RetryCallableWrapper(Runnable targetTask, RetryTemplate retryTemplate) {this.targetTask = targetTask;this.retryTemplate = retryTemplate;}@Overridepublic void run() {retryTemplate.execute(context -> {targetTask.run(); // 執行原始任務return null; // 返回類型無關緊要（Void）});}}/*** 上下文傳遞裝飾器：用于傳遞 MDC 上下文（如日志追蹤 ID）* - 通過 TaskDecorator 在任務執行前后傳遞上下文* - 確保異步任務中能訪問到主線程的上下文*/public static class MdcTaskDecorator implements TaskDecorator {@Overridepublic Runnable decorate(Runnable runnable) {// 獲取當前線程的 MDC 上下文（如 traceId）Map<String, String> contextMap = MDC.getCopyOfContextMap();return () -> {try {// 恢復上下文（如 traceId）MDC.setContextMap(contextMap);// 執行任務runnable.run();} finally {// 清理上下文（可選）MDC.clear();}};}}/*** 使用示例：提交任務到線程池* - 演示上下文傳遞和重試機制*/public void useTaskExecutor() {ThreadPoolTaskExecutor executor = taskExecutor();// 示例1：提交帶重試的任務Runnable retryableTask = new RetryCallableWrapper(() -> {System.out.println(Thread.currentThread().getName() + " 執行任務");throw new RuntimeException("模擬任務失敗"); // 模擬失敗}, createRetryTemplate());executor.execute(retryableTask); // 自動重試3次// 示例2：提交任務并傳遞上下文（如 MDC 日志追蹤 ID）MDC.put("traceId", "123456"); // 設置日志追蹤 IDexecutor.execute(() -> {System.out.println("當前 traceId: " + MDC.get("traceId")); // 輸出 123456System.out.println(Thread.currentThread().getName() + " 執行任務");});// 關閉線程池（Spring 容器會自動關閉）// executor.shutdown();}
}

---

3. 配置項詳解

3.1 線程池核心配置

配置項	說明	推薦值/示例
corePoolSize	核心線程數	CPU密集型：Runtime.getRuntime().availableProcessors()；IO密集型：2 * CPU核心數
maxPoolSize	最大線程數	根據業務負載調整（如 2 * corePoolSize）
keepAliveSeconds	空閑線程存活時間	與任務平均執行時間匹配（如 60 秒）
queue	任務隊列	ArrayBlockingQueue(25)（有界隊列）或 LinkedBlockingQueue（無界隊列）
rejectedExecutionHandler	拒絕策略	CallerRunsPolicy（調用線程執行）、AbortPolicy（拋異常）、DiscardPolicy（丟棄任務）
threadNamePrefix	線程名稱前綴	便于日志追蹤（如 "TaskExecutor-"）

3.2 上下文傳遞機制

• MDC（Mapped Diagnostic Context）：SLF4J 提供的日志上下文工具，用于記錄日志追蹤 ID。
• TaskDecorator：Spring 提供的裝飾器接口，用于在任務執行前后傳遞上下文。
• 實現方式：
  1. 在主線程中設置 MDC 上下文（如 MDC.put("traceId", "123456")）。
  2. 通過 TaskDecorator 將上下文傳遞給異步任務。
  3. 任務執行時恢復上下文，確保日志可追蹤。

3.3 重試機制

• RetryTemplate：Spring Retry 提供的重試模板，封裝了重試邏輯。
• SimpleRetryPolicy：定義最大重試次數（如 3 次）。
• FixedBackOffPolicy：定義重試間隔（如 1 秒）。
• 使用方式：
  1. 將任務包裝為 RetryCallableWrapper。
  2. 通過 RetryTemplate 執行任務，自動處理重試邏輯。

---

4. 使用場景建議

1. 日志追蹤：通過 MDC 傳遞 traceId，確保異步任務日志與主線程關聯。
2. 任務重試：適用于網絡請求、數據庫操作等需要自動重試的場景。
3. 資源控制：通過隊列和線程數限制，防止系統過載。

---

5. 注意事項

1. 重試邏輯與異常處理：
   • 默認對所有異常進行重試，可通過 SimpleRetryPolicy 自定義異常類型。
   • 重試后仍失敗的任務需通過日志或監控告警處理。
2. 上下文傳遞的開銷：
   • 如果上下文較大（如包含復雜對象），需評估性能影響。
   • 使用 InheritableThreadLocal 替代 MDC 可以更靈活地傳遞上下文。
3. 線程池生命周期：
   • 確保在應用關閉時正確關閉線程池（調用 executor.shutdown()）。
   • Spring 容器會自動關閉線程池，但手動創建時需顯式調用。
4. 隊列容量與拒絕策略：
   • 有界隊列需合理設置容量，防止任務堆積。
   • 拒絕策略應根據業務需求選擇（如 CallerRunsPolicy 適合輕量任務）。

---

6. 擴展功能（可選）

1. 動態調整線程池參數：通過 ThreadPoolTaskExecutor 提供的方法（如 setCorePoolSize()）動態修改配置。
2. 監控與調優：通過 ThreadPoolTaskExecutor 的統計方法（如 getPoolSize()、getQueueSize()）監控線程池狀態。
3. 自定義重試策略：結合 Spring Retry 實現更復雜的重試邏輯（如指數退避、重試條件判斷）。

---

7. 代碼結構圖

TaskExecutorConfig
├── taskExecutor()        // 配置線程池
│   ├── corePoolSize      // 核心線程數
│   ├── maxPoolSize       // 最大線程數
│   ├── keepAliveSeconds  // 空閑線程存活時間
│   ├── queue             // 任務隊列
│   ├── rejectedHandler   // 拒絕策略
│   └── setTaskDecorator  // 上下文傳遞機制
├── createRetryTemplate() // 創建重試模板
├── RetryCallableWrapper // 重試任務包裝器
└── useTaskExecutor()     // 使用示例├── execute()         // 提交普通任務└── submitListenable()// 提交帶返回值任務

---

8. 總結

• 上下文傳遞：確保異步任務中能訪問到主線程的上下文（如日志追蹤 ID）。
• 任務重試：對失敗任務自動重試，提升系統魯棒性。
• 靈活的線程池管理：根據業務需求調整線程池參數，避免資源浪費或過載。