2026/05/09/8 分钟/7 阅读/0 评论

Creeks 系统线程池与多线程使用分析

整个系统使用了 **3 种多线程模式**，涉及 **4 个独立的线程池配置类**，覆盖 **定时任务、并行查询、异步通知、报告处理** 等场景。假设服务器 8 核 CPU： Amazon 同步服务选是明智的——调用者自己跑任务，自动降速

分析日期：2026-05-09

一、整体概览

整个系统使用了 3 种多线程模式，涉及 4 个独立的线程池配置类，覆盖 定时任务、并行查询、异步通知、报告处理 等场景。

graph TB
    subgraph 线程池创建
    A[Spring @Async] -->|最简单| B[方法标注解, Spring自动路由到线程池]
    C[CompletableFuture.supplyAsync] -->|最灵活| D[手动创建任务, 指定线程池, join合并]
    E[CompletableFuture.runAsync] -->|最危险| F[不指定线程池, 用ForkJoinPool.commonPool]
    end

    subgraph 四个服务的线程池
    G[creeks-allocation<br/>jobTaskExecutor + completableFutureTaskExecutor]
    H[creekshub-sync-amazon<br/>esyncAmazonExecutor + reportHandlerExecutor + itemExtendHandlerExecutor + ...]
    I[creeks-sales<br/>goodsInventoryReportingExecutor + ebayInventoryReportingExecutor]
    J[codo-error-collect<br/>notifyExecutor]
    end

    subgraph 业务使用场景
    K[定时任务异步化]
    L[批量查询并行优化]
    M[异步告警通知]
    N[第三方API限频串行化]
    end

    B --> K
    D --> L
    B --> M
    D --> N

二、线程池创建方式：三种流派

流派	使用方式	优点	缺点	使用频率
@Async 注解驱动	方法上加 `@Async("executorName")`	最简单，一行注解	不支持返回值聚合	最普遍
CompletableFuture + 指定 Executor	`CompletableFuture.supplyAsync(runnable, executor)`	灵活，支持链式编排、结果聚合	需要手动写合并逻辑	较多
裸 CompletableFuture.runAsync()	`CompletableFuture.runAsync(runnable)`	随手写，最快	用 `ForkJoinPool.commonPool()`，无隔离	少（有隐患）

三、各服务线程池配置详解

3.1 creeks-allocation（调拨服务）—— 最完善

路径：E:\code\creeks-allocation\creeks-allocation-server\src\main\java\com\codo\tech\creeks\allocation\config\ThreadPoolTaskConfig.java

开启注解：@EnableAsync

两个线程池

配置项	`jobTaskExecutor`	`completableFutureTaskExecutor`
核心线程数	`CPU * 2 + 1`	`CPU * 2 + 1`
最大线程数	`min(核心*4, 96)`	`min(核心*4, 96)`
空闲存活时间	10 秒	10 秒
队列容量	50	未设置（默认 Integer.MAX_VALUE）
拒绝策略	`AbortPolicy`（抛异常）	`AbortPolicy`（抛异常）
优雅关闭	等待 60 秒	等待 60 秒
线程名前缀	`job-Async-Service-`	`completable-future-Async-Service-`

用途分工：

线程池	使用方式	典型场景
`jobTaskExecutor`	`@Async("jobTaskExecutor")`	定时任务异步化：拣货标签打印、可出库数量计算、履约包关系校验
`completableFutureTaskExecutor`	`CompletableFuture.supplyAsync(..., executor)`	批量查询并行化：Booking 详情列表、Shipment 详情列表

参数说明

假设服务器 8 核 CPU：

核心线程数 = 8 * 2 + 1 = 17
最大线程数 = min(17 * 4, 96) = 68

IO 密集型任务（查 DB、调远程接口）用 CPU * 2 + 1 是合理的选择。

3.2 creekshub-sync-amazon（Amazon 同步服务）—— 最"精打细算"

路径：E:\code\creekshub-sync-amazon\server\src\main\java\com\codo\tech\creekshub\sync\amazon\server\config\ThreadPoolTaskConfig.java

开启注解：@EnableAsync

五个线程池（按用途精细拆分）

Bean 名	核心线程	最大线程	队列	允许核心超时	拒绝策略	用途
`esyncAmazonExecutor`	CPU*2+1	min(*4, 96)	200	否	CallerRunsPolicy	通用异步任务
`reportHandlerExecutor`	5	5（默认）	无限	是	-	报告处理（少量几个线程，避免系统压力）
`itemExtendHandlerExecutor`	1	1（默认）	无限	是	-	商品扩展信息同步（单线程串行）
`inventoryReportHandlerExecutor`	1	1（默认）	无限	是	-	库存报告（ECPP 限频 360次/分钟）
`advertiseReportHandlerExecutor`	1	1（默认）	无限	是	-	广告报告（单线程串行）

这个服务很务实： 知道第三方 API 有频率限制（如 ECPP 360次/分钟），主动把并发压到 1 个线程，比盲目开多线程然后被限频强得多。

CallerRunsPolicy vs AbortPolicy

策略	行为	适用场景
`AbortPolicy`	队列满时直接抛 `RejectedExecutionException`	允许快速失败的场景
`CallerRunsPolicy`	队列满时由调用线程自己执行任务	不能丢任务的场景，自动给生产者"踩刹车"

Amazon 同步服务选 CallerRunsPolicy 是明智的——调用者自己跑任务，自动降速。

@Async 使用情况

类	方法	指定 Executor	风险
`AmazonFbaReturnRepository`	`fixReturnDateAt()`	否	用 Spring 默认 SimpleAsyncTaskExecutor，每次新建线程
`ItemExtendTask`	`syncSpItemExtend()`	`itemExtendHandlerExecutor`	安全
`ItemMarkNotFoundTask`	`markItemNotFound()`	`itemExtendHandlerExecutor`	安全
`FetchSpFinanceGroupTask`	`run()`	否	同上风险
`FetchAdvertiseReportTask`	`createAdvertiseReport()`	否	同上风险
`FetchAmazonAdvertiseInvoiceTask`	`startFetchAmazonAdvertiseInvoice()`	否	同上风险

3.3 creeks-sales（销售服务）—— 专注库存上报

路径：E:\code\creeks-sales\server\src\main\java\com\codo\tech\creeks\sales\config\ThreadPoolTaskConfig.java

开启注解：@EnableAsync

Bean 名	核心线程	最大线程	队列	拒绝策略	用途
`goodsInventoryReportingExecutor`	4	min(CPU*8, 96)	300	CallerRunsPolicy	商品可售数量上报（TikTokShop/Shopify 等）
`ebayInventoryReportingExecutor`	10	15	50	CallerRunsPolicy	eBay 库存上报

设计思路： 库存上报是 IO 密集型（调电商平台 API），eBay 给的并发数更多所以核心线程 10，TikTokShop/Shopify 限频严所以核心线程 4。队列小（50）追求快速响应，满了由调用者自己跑。

@Async 使用情况

类	方法	指定 Executor	说明
`TiktokShopGoodsReportingInventoryTask`	`processByStore()`	`goodsInventoryReportingExecutor`	按店铺并发上报库存
`ShopifyGoodsReportingInventoryTask`	`processByStore()`	`goodsInventoryReportingExecutor`	同上
`TiktokShopGoodsReportingRetryTask`	`start()`	否	重试任务
`TiktokShopGoodsInventoryHandleTask`	`start()`	否	库存处理入口
`ShopifyGoodsReportingInventoryTask`	`start()`	否	入口任务
`MarkBackendDeletedGoodsTask`	`start()`	否	标记删除任务
`GoodsJobServiceImpl`	`goodsSyncToES()`	否	商品同步 ES
`GoodsJobServiceImpl`	`goodsSyncToInventory()`	否	商品同步库存
`GoodsJobServiceImpl`	`handleSkuSaleAndStoreRelation()`	否	SKU 销售关系处理

大量 @Async 没指定 executor，用 Spring 默认 SimpleAsyncTaskExecutor（每次新建线程），有隐患。

3.4 codo-error-collect（错误收集服务）—— 最极简

路径：E:\code\codo-error-collect\server\src\main\java\com\codo\tech\error\collect\server\springconfig\AsyncConfig.java

notifyExecutor:
  corePoolSize = 1
  maxPoolSize  = 1
  queueCapacity = 1024
  threadNamePrefix = "NotifyThread-"

单线程 + 大队列，确保告警通知按顺序发，不会并发乱序。大队列（1024）保证短时间内大量错误日志进来不会丢弃。

四、多线程在业务中的使用模式

模式 1：@Async 定时任务异步化（最普遍）

典型代码：

@Async("jobTaskExecutor")
public void printPickingLabelInOrderJob() {
    lockService.lockAndExecute(JOB_CODE, 0, -1, TimeUnit.SECONDS, this::executeJob);
}

流程：

定时调度触发
    ↓
@Async 方法 → 扔给 jobTaskExecutor 线程池
    ↓
定时调度器立刻返回，不阻塞
    ↓
线程池中的线程执行实际任务

好处： 定时任务执行时间长不会阻塞下一次调度触发。

涉及场景：

服务	任务	线程池
allocation	拣货标签打印	jobTaskExecutor
allocation	可出库数量计算	jobTaskExecutor
allocation	履约包关系校验	默认（未指定）
sales	TikTokShop 库存上报	goodsInventoryReportingExecutor
sales	Shopify 库存上报	goodsInventoryReportingExecutor
sales	商品同步 ES	默认（未指定）
sales	商品同步库存	默认（未指定）
Amazon	财务组数据拉取	默认（未指定）
Amazon	广告报告拉取	默认（未指定）

模式 2：CompletableFuture 并行查询（allocation 用得最猛）

BookingBusinessImpl 批量查询详情页：

graph LR
    A[查询请求 N 条记录] --> B[按 10 条分批]
    B --> C1[批次1: CF.supplyAsync]
    B --> C2[批次2: CF.supplyAsync]
    B --> C3[批次N: CF.supplyAsync]
    C1 --> D[join 合并]
    C2 --> D
    C3 --> D
    D --> E[返回完整结果列表]

// 1. 分批，每批 10 条
int batchSize = 10;
List<List<Booking>> batches = Lists.partition(records, batchSize);

// 2. 每批一个 CompletableFuture 并行执行
List<CompletableFuture<List<BookingDetailVO>>> batchFutureList = Lists.newArrayList();
for (List<Booking> batch : batches) {
    CompletableFuture<List<BookingDetailVO>> batchFuture =
        CompletableFuture.supplyAsync(() -> {
            // 批次内部串行处理
            for (Booking record : batch) {
                BookingDetailVO vo = queryBookingBase(record, ...);  // 查 DB
                fillSaleSpecificInfo(vo);                              // 无 DB
                fillingShipmentTrack(vo);                              // 查 DB
                fillStockItemVoAndCartonStatisticsVo(vo, ...);         // 无 DB
            }
            return batchResults;
        }, completableFutureTaskExecutor);  // ← 指定专用线程池
    batchFutureList.add(batchFuture);
}

// 3. 合并结果
List<BookingDetailVO> result = batchFutureList.stream()
    .map(CompletableFuture::join)
    .flatMap(List::stream)
    .collect(Collectors.toList());

ShipmentBusinessImpl 同理：

for (ShipmentResponse response : orderList) {
    CompletableFuture<ShipmentDetailVo> singleFuture =
        CompletableFuture.supplyAsync(() -> {
            ShipmentDetailVo vo = buildShipmentDetailVo(response);
            fillingTrackInfo(vo);  // 查轨迹
            return vo;
        }, completableFutureTaskExecutor);
    completableFutureList.add(singleFuture);
}

优化效果： 假设 100 条记录，原来串行查 100 次 DB，现在分 10 批并行（每批 10 条串行），查询时间从 100 单位降到约 10 单位。

模式 3：裸 CompletableFuture.runAsync()（有隐患）

// InventoryAgeSpMxReportService.java:198
CompletableFuture.runAsync(() -> {
    try (InputStream in = Files.newInputStream(tempFile.toPath())) {
        // 上传文件到 OSS
    }
});

问题： 没指定线程池，用的是 ForkJoinPool.commonPool()。

commonPool() 的线程数 = CPU核心数 - 1，是 JVM 全局共享的，跟业务线程池没有隔离。如果多个地方都用 commonPool，会互相争抢。

涉及文件：

文件	行号	用途
`InventoryAgeSpMxReportService.java`	198	上传库存年龄报告到 OSS
`InventoryAgeSpReportService.java`	204	上传库存年龄报告到 OSS

五、FutureUtils 工具类

路径：E:\code\creeks-allocation\creeks-allocation-server\src\main\java\com\codo\tech\creeks\allocation\utils\FutureUtils.java

allocation 服务封装了一套 CompletableFuture 合并工具类：

方法	输入	输出	适用场景
`sequence()`	`List<CompletableFuture<T>>`	`CompletableFuture<List<T>>`	多个异步任务结果收集
`sequenceList()`	`List<CompletableFuture<List<T>>>`	`CompletableFuture<List<T>>`	分页查询场景（扁平化）
`sequenceMap()`	`List<CompletableFuture<Map<K,V>>>`	`CompletableFuture<Map<K,V>>`	多个 Map 合并，自定义 key 冲突策略
`sequenceNonNull()`	`List<CompletableFuture<T>>`	`CompletableFuture<List<T>>`	合并时自动过滤 null
`sequenceListNonNull()`	`List<CompletableFuture<List<T>>>`	`CompletableFuture<List<T>>`	分页查询 + 过滤 null
`extractRealException()`	`Throwable`	`Throwable`	从 CompletionException 中剥出真实异常

实际使用： 业务代码里用得不多，主要还是手写的 stream().map(CompletableFuture::join)。

六、完整使用矩阵

@Async 使用情况汇总

服务	类	方法	指定 Executor	Executor 名	风险等级
allocation	PrintPickingLabelInOrderJobServiceImpl	printPickingLabelInOrderJob()	是	jobTaskExecutor	安全
allocation	AllocationOrderBusinessImpl	calculateTotalCanOutboundQuantity()	是	jobTaskExecutor	安全
allocation	FulfillmentPackRelationCheckJobService	checkFulfillmentPackRelation()	否	默认	中
sales	TiktokShopGoodsReportingInventoryTask	processByStore()	是	goodsInventoryReportingExecutor	安全
sales	ShopifyGoodsReportingInventoryTask	processByStore()	是	goodsInventoryReportingExecutor	安全
sales	TiktokShopGoodsReportingRetryTask	start()	否	默认	中
sales	TiktokShopGoodsInventoryHandleTask	start()	否	默认	中
sales	ShopifyGoodsReportingInventoryTask	start()	否	默认	中
sales	MarkBackendDeletedGoodsTask	start()	否	默认	中
sales	GoodsJobServiceImpl	goodsSyncToES()	否	默认	中
sales	GoodsJobServiceImpl	goodsSyncToInventory()	否	默认	中
sales	GoodsJobServiceImpl	handleSkuSaleAndStoreRelation()	否	默认	中
Amazon	AmazonFbaReturnRepository	fixReturnDateAt()	否	默认	中
Amazon	ItemExtendTask	syncSpItemExtend()	是	itemExtendHandlerExecutor	安全
Amazon	ItemMarkNotFoundTask	markItemNotFound()	是	itemExtendHandlerExecutor	安全
Amazon	FetchSpFinanceGroupTask	run()	否	默认	中
Amazon	FetchAdvertiseReportTask	createAdvertiseReport()	否	默认	中
Amazon	FetchAmazonAdvertiseInvoiceTask	startFetchAmazonAdvertiseInvoice()	否	默认	中

CompletableFuture 使用情况汇总

服务	类	用途	指定 Executor	风险等级
allocation	BookingBusinessImpl	批量查询 Booking 详情	是 completableFutureTaskExecutor	安全
allocation	ShipmentBusinessImpl	批量查询 Shipment 详情	是 completableFutureTaskExecutor	安全
Amazon	InventoryAgeSpReportService	上传报告到 OSS	否 commonPool()	高
Amazon	InventoryAgeSpMxReportService	上传报告到 OSS	否 commonPool()	高

七、潜在问题与改进建议

7.1 @Async 不指定 Executor = 裸奔

问题： 约一半的 @Async 注解没有指定 executor 名称：

@Async  // ← 没指定！
public void run() { ... }

Spring 在没有指定 executor 时使用 SimpleAsyncTaskExecutor，每次调用创建新线程，没有池化，线程数无上限。高并发场景下能把 JVM 线程数打爆。

建议：

给所有 @Async 加上 executor 参数：@Async("esyncAmazonExecutor")
或者在 application.yml 中配置默认异步线程池：

spring:
  task:
    execution:
      pool:
        core-size: 10
        max-size: 50
        queue-capacity: 200

7.2 裸 CompletableFuture.runAsync() 用 commonPool

问题： InventoryAgeSpReportService 和 InventoryAgeSpMxReportService 里：

CompletableFuture.runAsync(() -> {
    // 上传文件到 OSS
});

ForkJoinPool.commonPool() 大小 = CPU核心数 - 1，JVM 全局共享，跟业务线程池没有隔离。

建议： 传入指定的 executor：

CompletableFuture.runAsync(() -> { ... }, reportHandlerExecutor);

7.3 completableFutureTaskExecutor 无队列上限

问题： allocation 的 completableFutureTaskExecutor 没调用 setQueueCapacity()，默认 Integer.MAX_VALUE。

// ThreadPoolTaskConfig.java:46-61
@Bean("completableFutureTaskExecutor")
public ThreadPoolTaskExecutor completableFutureTaskExecutor() {
    // ...
    // ← 没有 setQueueCapacity()！
    threadPoolTaskExecutor.initialize();
    return threadPoolTaskExecutor;
}

任务堆积时队列能撑爆内存，而且 AbortPolicy 拒绝策略在这种情况下永远触发不了（因为队列永远不会满）。

建议： 设置合理上限：

threadPoolTaskExecutor.setQueueCapacity(200);

7.4 线程池配置散落，没有统一基座

问题： 4 个服务各自定义 ThreadPoolTaskConfig，参数公式高度雷同：

allocation:       core = CPU*2+1,  max = min(*4,96), queue = 50
amazon-sync:      core = CPU*2+1,  max = min(*4,96), queue = 200
sales:            core = CPU*2,    max = min(*4,96), queue = 300

重复代码，维护成本大。

建议： 在 creeks-common-services 中提供 AbstractThreadPoolConfig 基类，各服务继承并覆盖差异化参数：

// creeks-common-services
public abstract class AbstractThreadPoolConfig {
    protected int defaultCoreSize() {
        return Runtime.getRuntime().availableProcessors() * 2 + 1;
    }
    protected int defaultMaxSize() {
        return Math.min(defaultCoreSize() * 4, 96);
    }
    // ...
}

7.5 多个线程池的拒绝策略选择

服务	线程池	拒绝策略	评价
allocation	jobTaskExecutor	AbortPolicy	合理，快速失败
allocation	completableFutureTaskExecutor	AbortPolicy	队列无上限，实际上不会触发
amazon-sync	esyncAmazonExecutor	CallerRunsPolicy	合理，不能丢任务
sales	goodsInventoryReportingExecutor	CallerRunsPolicy	合理
sales	ebayInventoryReportingExecutor	CallerRunsPolicy	合理
codo-error-collect	notifyExecutor	默认 AbortPolicy	合理

建议： CallerRunsPolicy 在不能丢任务的场景下是对的，但要注意"调用者自己跑"会阻塞调用方线程，如果调用方也是线程池中的线程，可能导致线程池饥饿。

八、关键文件索引

文件	路径	作用
ThreadPoolTaskConfig.java (allocation)	`creeks-allocation/.../config/`	调拨服务线程池配置（2个）
ThreadPoolTaskConfig.java (amazon)	`creekshub-sync-amazon/.../config/`	Amazon同步服务线程池配置（5个）
ThreadPoolTaskConfig.java (sales)	`creeks-sales/.../config/`	销售服务线程池配置（2个）
AsyncConfig.java	`codo-error-collect/.../springconfig/`	错误收集服务线程池配置（1个）
FutureUtils.java	`creeks-allocation/.../utils/`	CompletableFuture 合并工具类
BookingBusinessImpl.java	`creeks-allocation/.../business/impl/`	CompletableFuture 并行查询示例
ShipmentBusinessImpl.java	`creeks-allocation/.../business/impl/`	CompletableFuture 并行查询示例
InventoryAgeSpReportService.java	`creekshub-sync-amazon/.../report/service/`	裸 runAsync 隐患示例
InventoryAgeSpMxReportService.java	`creekshub-sync-amazon/.../report/service/`	裸 runAsync 隐患示例
ItemExtendTask.java	`creekshub-sync-amazon/.../report/service/`	@Async 指定 executor 示例
TiktokShopGoodsReportingInventoryTask.java	`creeks-sales/.../job/`	@Async 按店铺并行示例

Creeks 系统线程池与多线程使用分析

一、整体概览

二、线程池创建方式：三种流派

三、各服务线程池配置详解

3.1 creeks-allocation（调拨服务）—— 最完善

两个线程池

参数说明

3.2 creekshub-sync-amazon（Amazon 同步服务）—— 最"精打细算"

五个线程池（按用途精细拆分）

CallerRunsPolicy vs AbortPolicy

@Async 使用情况

3.3 creeks-sales（销售服务）—— 专注库存上报

@Async 使用情况

3.4 codo-error-collect（错误收集服务）—— 最极简

四、多线程在业务中的使用模式

模式 1：@Async 定时任务异步化（最普遍）

模式 2：CompletableFuture 并行查询（allocation 用得最猛）

模式 3：裸 CompletableFuture.runAsync()（有隐患）

五、FutureUtils 工具类

六、完整使用矩阵

@Async 使用情况汇总

CompletableFuture 使用情况汇总

七、潜在问题与改进建议

7.1 @Async 不指定 Executor = 裸奔

7.2 裸 CompletableFuture.runAsync() 用 commonPool

7.3 completableFutureTaskExecutor 无队列上限

7.4 线程池配置散落，没有统一基座

7.5 多个线程池的拒绝策略选择

八、关键文件索引

商品（Goods）、产品（SKU/SPU/SKC）、货物（Stock）关系与业务流程分析

多平台财务核算系统

多线程与JUC并发编程实战

评论 0