多线程与JUC并发编程实战

背景与痛点

业务背景

Creeks ERP 系统是一套跨境电商全链路管理平台，涵盖库存（inventory）、销售（sales）、履约（fulfillment）、财务（finance）、平台同步（esync）等十余个微服务模块。系统对接了 eBay、Amazon、Walmart、Shopify、TikTok Shop 等多个跨境电商平台，日常涉及大量的外部 API 调用、批量数据处理和异步任务调度。

核心痛点

1. 外部API调用慢且有限制：万邑通（Winit）物流商API单次请求限制1000PCS，eBay API单次最多修改4个商品且15秒内不超过6000次调用，串行处理严重拖慢业务流程
2. 派单吞吐量瓶颈：出库单派单涉及多个仓库类型（自营仓、万邑通仓），万邑通API响应慢（秒级到十几秒），串行派单导致其他类型订单被阻塞
3. 分布式环境下的并发安全：多实例部署下，定时任务重复执行、同一单据被并发操作、分布式事务脏读等问题频发

三个核心多线程场景

场景	模块	核心问题	JUC方案
万邑通标签请求拆解	inventory	API限制1000PCS，大请求需拆分并发	CountDownLatch + 数组保序 + 手动线程池
eBay可售数量上报	esync-ebay	8000W商品上报，API限制4个/批	CompletableFuture + ConcurrentLinkedQueue
多线程派单	inventory	万邑通慢请求阻塞其他派单	@Async + Future + 线程池隔离（舱壁模式）

---

技术方案

场景1：万邑通标签请求拆解（分治模式 + CountDownLatch）

架构设计

请求入口 (getWinitStockLabel)
    │
    ▼
Redis去重 + 幂等检查 (RBucket)
    │
    ▼
Kafka异步消息发送
    │
    ▼
消费者处理 (WinitLabelRequestConsumer, max.poll.records:1)
    │
    ▼
分布式锁 (LockService)
    │
    ▼
任务拆分 (WinitLabelRequestDivider, 阈值900)
    ├─ 输入：[99, 800, 120, 1555] 件
    └─ 输出：[900, 900, 674] 三个子请求
    │
    ▼
多线程并发请求 (CountDownLatch + ThreadPoolExecutor)
    ├─ 线程1: 请求900件 → resultArray[0]
    ├─ 线程2: 请求900件 → resultArray[1]
    └─ 线程3: 请求674件 → resultArray[2]
    │
    ▼
结果合并 (WinitLabelResponseCombiner)
    ├─ PDF文件合并（按数组索引顺序）
    ├─ S码列表合并
    └─ 映射关系建立 (StockLabelRelation)
    │
    ▼
上传七牛 + 数据库保存

核心代码分析

线程池配置（WinitLabelRequestThreadPool）— DCL单例模式：

private static final int DEFAULT_CORE_SIZE = 0;      // 核心线程数为0，空闲时不占资源
private static final int MAX_THREAD_SIZE = 100;       // 最大线程数100，支持高并发
private static final long KEEP_ALIVE_TIME = 2000L;    // 空闲线程2秒后回收
private static volatile ThreadPoolExecutor executor;   // volatile保证可见性

public static ThreadPoolExecutor getInstance() {
    if (executor == null) {
        synchronized (WinitLabelRequestThreadPool.class) {
            if (executor == null) {
                executor = new ThreadPoolExecutor(
                    DEFAULT_CORE_SIZE, MAX_THREAD_SIZE, KEEP_ALIVE_TIME,
                    TimeUnit.MILLISECONDS,
                    new SynchronousQueue<>(),           // 无缓冲队列，即时交付
                    Executors.defaultThreadFactory(),
                    new ThreadPoolExecutor.CallerRunsPolicy());  // 调用线程执行，天然背压
            }
        }
    }
    return executor;
}

多线程协调（WinitLabelRequestHelper.doRequest()）：

CountDownLatch countDownLatch = new CountDownLatch(requestDataList.size());
PrintSingleItemLabelResponse[] resultArray =
    new PrintSingleItemLabelResponse[requestDataList.size()];  // 数组按索引保序
StringBuffer errMsg = new StringBuffer();  // 线程安全的错误收集

for (int i = 0; i < resultArray.length; i++) {
    poolExecutor.execute(
        new WinitLabelRequestTask(data, context, witSdkProductService,
            countDownLatch, resultArray, errMsg, i));
}
countDownLatch.await();  // 阻塞等待所有子请求完成

任务执行体（WinitLabelRequestTask.run()）— 快速失败机制：

public void run() {
    try {
        // 已有错误时不继续执行，加快返回速度
        if (StrUtil.isBlank(errMsg.toString())) {
            BaseWinitResponse<PrintSingleItemLabelResponse> winitResponse =
                witSdkProductService.printSingleItemLabel(context, request);
            resArray[index] = responseData;  // 每个线程写不同索引，天然无竞争
        }
    } catch (Exception e) {
        errMsg.append(e.getMessage());  // 全局错误收集
        throw e;
    } finally {
        this.countDownLatch.countDown();  // 确保countDown一定执行
    }
}

JUC选型理由

组件	选型理由	替代方案对比
手动ThreadPoolExecutor（核心0/最大100/SynchronousQueue）	核心0空闲不占资源，SynchronousQueue无缓冲即时交付，适合突发性标签请求	如果用LinkedBlockingQueue，任务会堆积在队列中，无法立即创建新线程处理
CallerRunsPolicy	线程池满时由Kafka消费线程执行，天然背压，不丢任务	AbortPolicy会丢弃任务；DiscardPolicy静默丢弃更危险
CountDownLatch	等待所有子请求完成后才能合并PDF，一次性栅栏最简实现	CyclicBarrier可重用但这里不需要；CompletableFuture更优雅但当时选择了更直接的方案
数组按索引写入	PDF合并必须有序，每线程写不同索引位置天然无竞争，无需任何同步	ConcurrentLinkedQueue无法保序；Collections.synchronizedList有锁开销
DCL单例（volatile + synchronized）	线程池全局唯一，volatile防止指令重排序导致的半初始化问题	枚举单例更安全但不适合需要参数化构造的场景

---

场景2：eBay可售数量上报（CompletableFuture + ConcurrentLinkedQueue）

架构设计

上报任务触发 (EbayGoodsReportingInventoryTask)
    │
    ▼
按店铺维度异步处理（前20个优先级最高的店铺）
    │
    ▼
每店铺最多1000条Feed，每批100条
    │
    ▼
ReviseInventoryServiceImpl.reviseInventory()
    │
    ▼
ListUtils.partition(itemList, 4)  ← eBay API限制：单次最多4个商品
    │
    ▼
CompletableFuture.runAsync() × N  ← 提交到ebayInventoryReportingExecutor(50/75)
    │                                    │
    │  ┌─────────────────────────────────┘
    │  │
    ▼  ▼
handleReportInventory()  ← 每个任务创建独立ReviseInventoryStatusCall实例
    │
    ├─ 成功 → successQueue.add(items)     ← ConcurrentLinkedQueue无锁收集
    │
    └─ 失败 → 降级为逐个上报
         ├─ 单个成功 → successQueue.add(item)
         └─ 单个失败 → failQueue.add(item)  ← 记录失败原因
    │
    ▼
CompletableFuture.allOf().get()  ← 等待所有分区完成
    │
    ▼
合并结果：new ArrayList<>(successQueue) / new ArrayList<>(failQueue)

核心代码分析

完整的上报流程（ReviseInventoryServiceImpl）：

// 1. 按4个一组分区（eBay API限制）
List<List<ReportingInventoryItem>> itemListPartition = ListUtils.partition(itemList, 4);

// 2. 无锁并发队列收集结果
ConcurrentLinkedQueue<ReportingInventoryItem> successQueue = new ConcurrentLinkedQueue<>();
ConcurrentLinkedQueue<ReportingInventoryItem> failQueue = new ConcurrentLinkedQueue<>();

// 3. CompletableFuture异步提交
List<CompletableFuture<Void>> futures = new ArrayList<>();
for (List<ReportingInventoryItem> items : itemListPartition) {
    CompletableFuture<Void> future = CompletableFuture.runAsync(
        () -> handleReportInventory(successQueue, failQueue, apiContext, items),
        ebayInventoryReportingExecutor);
    futures.add(future);
}

// 4. 等待所有任务完成
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).get();

// 5. 合并结果
response.setSuccessList(new ArrayList<>(successQueue));
response.setFailList(new ArrayList<>(failQueue));

三层递进异常处理：

private void handleReportInventory(...) {
    // 为每个任务创建独立的调用实例，避免线程安全问题
    ReviseInventoryStatusCall call = new ReviseInventoryStatusCall(apiContext);
    InventoryStatusType[] inventoryStatus = items.stream()
        .map(this::convertToInventoryStatusType).toArray(InventoryStatusType[]::new);
    call.setInventoryStatus(inventoryStatus);

    try {
        call.reviseInventoryStatus();       // 第一层：批量4个上报
        successList.addAll(items);
    } catch (Exception e) {
        for (ReportingInventoryItem item : items) {  // 第二层：逐个重试
            call.setInventoryStatus(new InventoryStatusType[]{convertToInventoryStatusType(item)});
            try {
                call.reviseInventoryStatus();
                successList.add(item);
            } catch (Exception eWithOneItem) {       // 第三层：记录失败
                item.setMessage(eWithOneItem.getMessage());
                failList.add(item);
            }
        }
    }
}

JUC选型理由

组件	选型理由	替代方案对比
CompletableFuture.runAsync()	无返回值异步任务，比Future更现代，支持链式编排和异常传播	用@Async需要代理调用，增加复杂度
CompletableFuture.allOf()	一次性等待所有分区完成，比CountDownLatch更优雅，自带异常传播	CountDownLatch需要手动管理countDown
ConcurrentLinkedQueue	基于CAS的无锁队列，多线程并发add无竞争，O(1)性能	synchronizedList全局锁竞争严重；CopyOnWriteArrayList写时复制开销大
线程池核心50/最大75	eBay上报量大且持续，核心50保证基线并发度	核心数太小会频繁创建销毁线程

---

场景3：多线程派单（@Async + Future + 四层锁 + 舱壁模式）

架构设计

派单触发 (notifyPushWms)
    │
    ├─ lockAndDispatchOrder()          ← @Async("jobTaskExecutor") 非万邑通派单
    │       │
    │       ▼
    │   Redisson分布式锁 (NOTIFY_PUSH_WMS_LOCK_NAME)  ← 第一层：防止多实例重复
    │       │
    │       ▼
    │   dispatchOrder() while循环（最多300次）
    │       │
    │       ▼
    │   queryOrderPushRecord()         ← @Transactional + @GlobalLock + FOR UPDATE
    │       │                             第二层：Seata全局锁防脏读
    │       │                             第三层：FOR UPDATE行锁防重复查询
    │       ▼
    │   逐条 @Async("inventoryTaskExecutor") 提交
    │       │
    │       ▼
    │   push() → tryPush() → doPush()  ← @LockOrder 第四层：单据级锁
    │       │
    │       ▼
    │   Future.get() 循环等待所有结果
    │
    └─ lockAndDispatchWinitOrderOnly() ← @Async("jobTaskExecutor") 万邑通专用
            │
            ▼
        Redisson分布式锁 (NOTIFY_PUSH_WMS_WINIT_LOCK_NAME)
            │
            ▼
        dispatchWinitOrderOnly() → push2Winit()  ← @Async("winitTaskExecutor") 隔离慢请求

核心代码分析

三个线程池的隔离设计：

// 通用库存任务线程池 — 处理非万邑通派单
@Bean("inventoryTaskExecutor")
// 核心5 / 最大96 / 队列300 / CallerRunsPolicy

// Job定时任务线程池 — 处理定时调度
@Bean("jobTaskExecutor")
// 核心10 / 最大50 / 队列300 / AbortPolicy（宁可拒绝也不阻塞调度线程）

// 万邑通专用线程池 — 隔离慢请求（舱壁模式）
@Bean("winitTaskExecutor")
// 核心2 / 最大4 / 队列300 / CallerRunsPolicy
// 万邑通API响应慢（秒级~十几秒），小线程池既控制并发又隔离影响

派单主循环（dispatchOrder）：

private void dispatchOrder(Integer pushType) {
    int count = 0;
    Long minId = this.getMinId(pushType);
    while (true) {
        if (count > PUSH_WMS_PROCESS_LIMIT) {  // 300次上限，防止单次执行过长
            break;
        }
        // @GlobalLock + FOR UPDATE 查询100条待派单记录
        List<OrderPushRecord> orderPushRecordList =
            SpringContextUtils.getBean(this.getClass())
                .queryOrderPushRecord(pushType, minId, false);
        if (CollectionUtils.isEmpty(orderPushRecordList)) break;

        List<Future<String>> futureList = new ArrayList<>();
        orderPushRecordList.forEach(orderPushRecord -> {
            BaseOrder baseOrder = GsonTool.jsonToBean(reqContent, BaseOrder.class);
            // @Async提交到inventoryTaskExecutor
            futureList.add(SpringContextUtils.getBean(this.getClass())
                .push(orderPushRecord, baseOrder, pushType));
        });

        // 等待所有Future完成
        for (Future<String> future : futureList) {
            try {
                String orderNum = future.get();  // 阻塞等待
            } catch (Exception e) {
                log.error("{} 执行失败", future, e);  // 异常不中断，继续处理
            }
        }
        minId = orderPushRecordList.get(orderPushRecordList.size() - 1).getId();
        count++;
    }
}

四层锁机制详解：

// 第一层：Redisson分布式锁 — 防止多实例重复执行派单Job
lockService.lockAndExecute(
    RedisKeyConstant.NOTIFY_PUSH_WMS_LOCK_NAME + pushType,
    0, -1, TimeUnit.SECONDS,  // waitTime=0立即返回，leaseTime=-1自动续期
    () -> this.dispatchOrder(pushType));

// 第二层：Seata全局锁 — 解决OMS批量派单分布式事务脏读
@Transactional
@GlobalLock
public List<OrderPushRecord> queryOrderPushRecord(...) {
    // 第三层：FOR UPDATE行锁 — 防止同一批数据被重复查询
    queryWrapper.last("limit 100 FOR UPDATE");
    return invOrderPushRecordService.list(queryWrapper);
}

// 第四层：@LockOrder单据级锁 — 防止同一单据并发操作（如同时派单和取消）
@LockOrder(fieldOrderNo = "#baseOrder.outboundOrderNo", orderType = OrderTypeEnum.OUTBOUND_ORDER)
public String doPush(BaseOrder baseOrder, Integer pushType) {
    // 执行派单/取消/退回
}

JUC选型理由

组件	选型理由	替代方案对比
@Async + Future + AsyncResult	Spring原生异步，与事务管理和AOP无缝集成	CompletableFuture更灵活但需要手动管理线程池注入
三个独立线程池（舱壁模式）	万邑通API慢，隔离防止阻塞其他派单；定时任务用AbortPolicy防止堆积	共用线程池会导致万邑通慢请求占满线程，其他类型派单饿死
SpringContextUtils.getBean()	确保@Async和@Transactional的AOP代理生效	直接this调用会绕过代理，注解失效
Redisson分布式锁	支持自动续期（看门狗机制），防止任务执行时间超过锁过期时间	基于Redis SETNX的简单锁不支持续期

---

三场景JUC选型对比总览

维度	场景1：标签拆解	场景2：eBay上报	场景3：多线程派单
线程池	手动ThreadPoolExecutor（0/100/SynchronousQueue）	Spring ThreadPoolTaskExecutor（50/75/300）	三个Spring线程池（隔离）
同步机制	CountDownLatch	CompletableFuture.allOf()	Future.get()循环
结果收集	数组按索引（保序）	ConcurrentLinkedQueue（无序）	Future逐个获取
异常处理	StringBuffer全局收集 + 快速失败	三层递进（批量→单个→记录）	try-catch继续 + 重试计数
线程安全	数组索引隔离 + DCL单例	ConcurrentLinkedQueue + 独立实例	四层锁（分布式锁→行锁→全局锁→单据锁）
背压/限流	CallerRunsPolicy	线程池队列300	AbortPolicy + 300次循环上限

---

实现细节与踩坑点

1. CountDownLatch.await() 无超时 — 永久阻塞风险

WinitLabelRequestHelper.java:70 行的 countDownLatch.await() 没有设置超时时间。如果万邑通API hang住或网络异常导致某个线程永远不执行 countDown()，主线程将永久阻塞，Kafka消费线程被占用，后续消息无法消费。

// 当前代码（有风险）
countDownLatch.await();

// 改进方案
boolean completed = countDownLatch.await(30, TimeUnit.SECONDS);
if (!completed) {
    throw new BusinessException(500, "万邑通标签请求超时，未完成的任务数：" + countDownLatch.getCount());
}

2. StringBuffer 的过度使用

WinitLabelRequestHelper 用 StringBuffer 收集多线程错误信息，但实际上只需要知道"是否有错误"。StringBuffer 的 synchronized 方法级锁在高并发下有不必要的性能开销。

// 当前代码
StringBuffer errMsg = new StringBuffer();
// 检查时：StrUtil.isNotBlank(errMsg.toString())  ← 每次toString()都创建新String

// 改进方案：用AtomicReference只记录第一个异常
AtomicReference<Throwable> firstError = new AtomicReference<>();
// 在任务中：firstError.compareAndSet(null, e);  ← CAS无锁，只记录第一个
// 检查时：if (firstError.get() != null) throw ...

3. @PostConstruct + 静态变量注入Spring Bean — 反模式

WinitLabelRequestHelper 通过 @PostConstruct 将Spring Bean赋值给静态变量，然后在静态方法 doRequest() 中使用。这种方式破坏了Spring的依赖注入机制，不利于单元测试和维护。

// 当前代码（反模式）
private static WinitLabelRequestHelper helper;
@PostConstruct
public void init() {
    helper = this;
    helper.witSdkProductService = this.witSdkProductService;
}
public static List<PrintSingleItemLabelResponse> doRequest(...) {
    // 使用 helper.witSdkProductService
}

// 改进方案：改为实例方法，通过Spring注入调用
@Resource private WinitLabelRequestHelper winitLabelRequestHelper;
winitLabelRequestHelper.doRequest(...);  // 实例方法调用

4. Future.get() 无超时 + 串行等待

OutboundJobServiceImpl.java:288 行的 future.get() 没有超时，且是串行等待。假设有10个任务，第1个耗时10秒，即使第2-10个已完成，也要等第1个返回后才能处理后续结果。

// 当前代码（有风险）
for (Future<String> future : futureList) {
    String orderNum = future.get();  // 无超时，串行等待
}

// 改进方案A：加超时
String orderNum = future.get(30, TimeUnit.SECONDS);

// 改进方案B：用CompletionService按完成顺序获取
ExecutorCompletionService<String> completionService = new ExecutorCompletionService<>(executor);
// 提交任务...
for (int i = 0; i < taskCount; i++) {
    Future<String> completed = completionService.take();  // 按完成顺序获取
    String orderNum = completed.get();
}

5. dispatchOrder 和 dispatchWinitOrderOnly 代码重复

OutboundJobServiceImpl.java 第252-298行和第300-346行，两个方法逻辑几乎完全相同，只是线程池和查询条件不同。违反DRY原则。

// 改进方案：抽取模板方法
private void doDispatch(Integer pushType, Boolean isWinit,
                        BiFunction<OrderPushRecord, BaseOrder, Future<String>> pushFunc) {
    int count = 0;
    Long minId = this.getMinId(pushType, isWinit);
    while (true) {
        if (count > PUSH_WMS_PROCESS_LIMIT) break;
        List<OrderPushRecord> records = proxy().queryOrderPushRecord(pushType, minId, isWinit);
        if (CollectionUtils.isEmpty(records)) break;
        List<Future<String>> futures = new ArrayList<>();
        records.forEach(record -> {
            BaseOrder baseOrder = GsonTool.jsonToBean(record.getReqContent(), BaseOrder.class);
            futures.add(pushFunc.apply(record, baseOrder));
        });
        waitForFutures(futures);
        minId = records.get(records.size() - 1).getId();
        count++;
    }
}

6. eBay上报单个重试时复用call对象

ReviseInventoryServiceImpl.java:108 行，批量失败降级为单个重试时，复用了同一个 ReviseInventoryStatusCall 实例。虽然在catch块内是串行执行的，但eBay SDK的Call对象内部可能有状态残留（如上次请求的响应数据），存在隐患。

// 当前代码（有隐患）
ReviseInventoryStatusCall call = new ReviseInventoryStatusCall(apiContext);
try {
    call.reviseInventoryStatus();  // 批量4个
} catch (Exception e) {
    for (ReportingInventoryItem item : items) {
        call.setInventoryStatus(...);  // 复用同一个call
        call.reviseInventoryStatus();  // 单个重试
    }
}

// 改进方案：每次重试创建新实例
for (ReportingInventoryItem item : items) {
    ReviseInventoryStatusCall singleCall = new ReviseInventoryStatusCall(apiContext);
    singleCall.setInventoryStatus(...);
    singleCall.reviseInventoryStatus();
}

7. 线程池核心线程数50偏大

ebayInventoryReportingExecutor 核心线程50，即使没有上报任务，也会保持50个线程存活。如果上报任务不是持续性的（比如每小时触发一次），空闲期间浪费资源。

// 改进方案：允许核心线程超时回收
taskExecutor.setAllowCoreThreadTimeOut(true);
taskExecutor.setKeepAliveSeconds(60);  // 60秒无任务则回收

---

多线程优化点挖掘

基于对整个ERP系统的分析，以下场景可以进一步引入多线程/并发优化：

优化点1：多平台库存同步并行化

esync模块对接了eBay/Amazon/Walmart/Shopee/Temu/TikTok等十几个平台。当库存变动需要同步到多个平台时，可以并行执行，每个平台独立线程池防止某平台API慢拖垮其他平台（舱壁模式）。

Map<Platform, CompletableFuture<SyncResult>> futures = platforms.stream()
    .collect(Collectors.toMap(Function.identity(),
        p -> CompletableFuture.supplyAsync(
            () -> syncInventory(p, skuList), platformExecutorMap.get(p))
            .orTimeout(30, TimeUnit.SECONDS)));  // Java 9+ 超时控制
CompletableFuture.allOf(futures.values().toArray(new CompletableFuture[0])).join();

优化点2：财务报表并行计算

系统已有 ParallelProcessingHelper 和 ConcurrentTaskSupport。利润计算中采购成本、头程成本、平台佣金、物流费用可以并行查询后汇总。

CompletableFuture<BigDecimal> purchaseCost = CompletableFuture.supplyAsync(() -> calcPurchaseCost(orderId));
CompletableFuture<BigDecimal> shippingCost = CompletableFuture.supplyAsync(() -> calcShippingCost(orderId));
CompletableFuture<BigDecimal> commission = CompletableFuture.supplyAsync(() -> calcCommission(orderId));
BigDecimal profit = revenue
    .subtract(purchaseCost.join())
    .subtract(shippingCost.join())
    .subtract(commission.join());

优化点3：物流轨迹批量查询

物流跟踪需要调用多个物流商API查询轨迹，天然适合并行。按物流商分组，组内并行但受Semaphore控制并发数（各物流商有API频率限制）。

Semaphore semaphore = new Semaphore(5);  // 每个物流商最多5个并发
CompletableFuture.runAsync(() -> {
    semaphore.acquire();
    try { queryTracking(carrier, trackingNo); }
    finally { semaphore.release(); }
}, carrierExecutor);

优化点4：大数据量导出异步化

报表导出数据量大时同步处理慢。可以用 @Async 异步生成 + Redis存储进度 + WebSocket/轮询通知完成。分页查询用 ConcurrentTaskSupport 并行拉取数据。

优化点5：订单批量处理流水线（生产者-消费者）

当前派单逐条 @Async 提交，可以优化为 BlockingQueue 流水线，解耦查询和处理，消费者可按需扩缩。

优化点6：缓存预热并行化

系统启动时需要加载大量基础数据（仓库信息、SKU映射、平台账号等），可以用 CompletableFuture.allOf() 并行预热多个缓存，缩短启动时间。

---

面试题与回答指南

Q1：你的万邑通标签线程池为什么核心线程数设为0？SynchronousQueue + CallerRunsPolicy组合有什么效果？

考察点：ThreadPoolExecutor参数含义、SynchronousQueue特性、拒绝策略的实际效果

参考回答：

核心线程数设为0，是因为标签请求是突发性的、非持续性的。大部分时间没有标签请求，如果保持核心线程存活会浪费资源。设为0意味着空闲时线程池没有任何线程，有任务时按需创建，2秒空闲后自动回收。

SynchronousQueue是一个不存储元素的队列，每个put操作必须等待一个take操作。这意味着任务提交时，如果有空闲线程就直接交给它，没有就创建新线程（直到最大100）。不会出现任务在队列中排队等待的情况，保证了任务的即时处理。

当100个线程都在忙时，新任务触发CallerRunsPolicy，由提交任务的线程（即Kafka消费线程）自己执行。效果是：

1. 不会丢弃任务
2. 自动降速——消费线程被占用期间不会消费新的Kafka消息，形成天然的背压机制

潜在风险：如果万邑通API持续慢响应，100个线程全部阻塞，CallerRunsPolicy导致Kafka消费线程也被阻塞，可能触发Kafka消费者rebalance。所以建议给CountDownLatch加超时。

---

Q2：场景1用CountDownLatch，场景2用CompletableFuture.allOf()，两者都是等待多个任务完成，为什么选型不同？

考察点：CountDownLatch和CompletableFuture的区别、异常传播机制

参考回答：

两者的核心区别在于异常传播和代码简洁度。

CountDownLatch：

• 一次性使用，手动countDown
• 不支持异常传播——子线程抛异常后，主线程的await()不会感知到，必须通过共享变量（如StringBuffer）传递错误信息
• 不支持超时（除非用await(timeout)重载）
• 优势：概念简单直接，适合"等所有任务完成"的纯同步场景

CompletableFuture.allOf()：

• 支持链式编排（thenApply/thenCombine/exceptionally）
• 自带异常传播——任何一个Future异常，allOf().get()会抛出ExecutionException
• 支持超时控制（orTimeout/completeOnTimeout，Java 9+）
• 优势：更现代，适合需要结果组合或异常处理的场景

场景1选CountDownLatch是因为：PDF合并需要按数组索引保序，用数组+CountDownLatch是最直接的实现。如果用CompletableFuture改造，可以用futures.get(i).join()按顺序获取结果，代码量减少约40%，异常处理也更清晰。

场景2选CompletableFuture是因为：结果不需要保序（成功/失败分别收集到ConcurrentLinkedQueue），且需要更优雅的异步编排。

---

Q3：eBay上报为什么选ConcurrentLinkedQueue而不是其他线程安全集合？

考察点：并发集合的内部实现和适用场景

参考回答：

这里的需求是：多个线程并发地往集合中add结果，最后主线程一次性读取。典型的"多生产者、单消费者"场景。

ConcurrentLinkedQueue：基于CAS的无锁队列，add操作O(1)，高并发下性能最好。没有锁竞争，不会阻塞。

Collections.synchronizedList：全局synchronized锁，每次add都要获取锁。50个线程并发add时，锁竞争严重，性能差。

CopyOnWriteArrayList：每次写操作复制整个底层数组。这里是写多读少（每个任务都add，最后才读一次），每次add都复制数组，性能极差且内存开销大。

LinkedBlockingQueue：支持阻塞等待（put/take），但这里不需要阻塞语义——生产者不需要等消费者，消费者也不需要等生产者。额外的ReentrantLock开销是浪费。

最终通过 new ArrayList<>(successQueue) 将结果转为普通List返回，ConcurrentLinkedQueue只在并发写入阶段使用。

---

Q4：派单场景用了四层锁（Redisson分布式锁、FOR UPDATE行锁、@GlobalLock、@LockOrder），每层解决什么问题？去掉某一层会怎样？

考察点：分布式锁、数据库锁、Seata全局锁的协作

参考回答：

第一层：Redisson分布式锁（lockAndExecute）

• 解决：防止多个服务实例同时执行派单Job
• 去掉后果：多实例重复查询同一批数据，导致重复派单

第二层：@GlobalLock（Seata全局锁）

• 解决：OMS批量派单走Seata全局事务，inventory查询时需要@GlobalLock确保读到的是已提交数据，防止分布式事务脏读
• 去掉后果：可能读到OMS尚未提交的中间状态数据，比如订单还在创建中就被派单

第三层：FOR UPDATE行锁

• 解决：防止同一批次内的数据被其他事务读取（比如另一个线程也在查询待派单记录）
• 去掉后果：并发查询可能读到相同记录，导致同一单据被多次提交派单

第四层：@LockOrder单据级锁

• 解决：防止同一单据被并发操作（如同时派单和取消）
• 去掉后果：同一出库单可能同时被派单和取消，状态不一致

四层锁从粗到细：集群级 → 事务级 → 行级 → 单据级，每层解决不同粒度的并发问题。

---

Q5：为什么万邑通派单要用独立的小线程池（2-4线程）？jobTaskExecutor为什么用AbortPolicy而不是CallerRunsPolicy？

考察点：线程池隔离（舱壁模式）、拒绝策略选择

参考回答：

万邑通独立线程池（舱壁模式）：

万邑通是跨境物流商，API响应时间通常在秒级到十几秒，远慢于自营仓库的毫秒级响应。如果万邑通派单和普通派单共用 inventoryTaskExecutor（最大96线程），万邑通的慢请求会逐渐占满线程池，导致普通派单也无法执行——这就是"线程池饥饿"问题。

用独立的 winitTaskExecutor（2-4线程）隔离万邑通请求，即使万邑通API全部超时，最多也只占用4个线程，不影响其他类型派单。这是微服务架构中"舱壁模式"（Bulkhead Pattern）的实践——就像船舱的隔水板，一个舱进水不会导致整艘船沉没。

线程数设为2-4而不是更多，是因为万邑通API本身有调用频率限制，线程太多反而会触发限流。

jobTaskExecutor用AbortPolicy：

定时任务场景下，如果线程池满说明上一批任务还没处理完。用CallerRunsPolicy会导致定时调度线程被占用，影响其他定时任务的触发时间。用AbortPolicy直接拒绝（抛RejectedExecutionException），被catch后记录日志，等下一次调度周期再处理。宁可这一轮少处理一些，也不能阻塞调度线程。

---

Q6：eBay上报的"批量4个→逐个重试→记录失败"三层递进异常处理有什么优势？如果直接逐个上报会怎样？

考察点：批量操作的异常处理策略、API调用优化

参考回答：

三层递进的优势：

正常情况下批量4个调用，API调用次数是 N/4。只有失败时才降级为逐个调用，最大化了正常路径的效率。

批量失败可能只是其中1个item有问题（比如listing已下架），逐个重试可以精确定位失败的item，其他3个仍然成功上报。这比直接整批标记失败要好得多。

如果直接逐个上报：

API调用次数变为N，是批量的4倍。eBay有"15秒内不超过6000次/卖家"的限制，逐个调用更容易触发限流。假设有2000个商品需要上报，批量模式只需500次调用，逐个模式需要2000次，在15秒窗口内的压力完全不同。

改进空间：可以在逐个重试前先检查批量失败的异常类型。如果是网络超时等非业务异常，可以直接整批重试而不是拆开——因为网络超时大概率4个都会失败，拆开重试只是浪费4倍的调用次数。

---

Q7：派单场景中 @Async + Future.get() 循环等待有什么问题？如何改进？

考察点：Future.get()的阻塞特性、串行等待问题

参考回答：

问题1：无超时
future.get() 没有超时参数，如果某个派单任务hang住（比如调用仓库API无响应），整个循环阻塞，后续所有单据都无法处理。

问题2：串行等待
循环中 future.get() 是按提交顺序串行等待的。假设有10个任务，第1个耗时10秒，第2-10个都在1秒内完成。主线程必须等第1个get()返回后才能处理第2个的结果，白白浪费了9秒。

问题3：异常处理
当前代码catch了异常后只是log.error，后续Future仍会继续get()，这是正确的。但如果有人改为抛出异常，后续Future的结果就丢失了。

改进方案：

方案A（最小改动）：加超时

方案B（推荐）：改用CompletableFuture + allOf

方案C：使用CompletionService按完成顺序获取

String orderNum = future.get(30, TimeUnit.SECONDS);

List<CompletableFuture<String>> futures = records.stream()
    .map(record -> CompletableFuture.supplyAsync(() -> doPush(record), executor))
    .collect(Collectors.toList());
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
    .orTimeout(60, TimeUnit.SECONDS).join();

ExecutorCompletionService<String> cs = new ExecutorCompletionService<>(executor);
// 提交任务...
for (int i = 0; i < taskCount; i++) {
    Future<String> completed = cs.take();  // 谁先完成谁先返回
    String orderNum = completed.get();
}

---

Q8：系统中出现了三种多线程结果收集方式：数组按索引、ConcurrentLinkedQueue、Future.get()。请对比线程安全原理和适用场景。

考察点：线程安全的不同实现方式

参考回答：

数组按索引写入（场景1）：

• 原理：每个线程写不同的索引位置（不同内存地址），天然无竞争，无需任何同步机制
• 适用：结果需要保序（如PDF合并），且任务数量固定已知
• 性能：最优，零同步开销
• 限制：必须预先知道任务数量来创建数组

ConcurrentLinkedQueue（场景2）：

• 原理：基于CAS（Compare-And-Swap）的无锁算法，内部用链表实现，add操作通过CAS更新tail指针
• 适用：结果不需要保序，任务数量可能动态变化，需要分类收集（如成功/失败分开）
• 性能：优秀，CAS在低竞争下接近无锁
• 限制：无界队列，极端情况下可能OOM

Future.get()逐个获取（场景3）：

• 原理：主线程串行调用每个Future的get()方法，不存在并发写入问题
• 适用：需要逐个处理结果（如记录日志），或者结果之间有依赖关系
• 性能：最差，串行等待开销最大
• 限制：按提交顺序等待，不是按完成顺序

性能排序：数组 > ConcurrentLinkedQueue > Future.get()

---

Q9：万邑通标签请求通过Kafka消费触发，派单通过定时任务触发。在分布式部署下，如何保证幂等性？Redis分布式锁失效会怎样？

考察点：幂等性设计、分布式锁的局限性

参考回答：

系统中的幂等手段：

标签请求：

• Kafka消费：max.poll.records:1 单条消费 + 手动ack（ack.acknowledge()），消费失败不确认，下次重新消费
• Redis幂等：用 RBucket 记录请求缓存key，重复请求直接返回
• 分布式锁：LockService.lockAndGetResult() 防止同一请求并发处理

派单：

• OrderPushRecord 表记录推送状态（NOT_PUSH/PUSHED），查询时过滤已推送记录
• retryTimes < 10 防止无限重试
• Redisson tryLock(0, -1) 防止多实例并发执行

Redis主从切换风险：

Redisson默认使用Redis单节点锁。主从切换时可能出现脑裂——主节点写入锁成功后宕机，从节点提升为主但没有这把锁，另一个实例也能获取到锁。后果是两个实例同时执行派单。

防御措施：

• 数据库层面：FOR UPDATE 行锁是最后一道防线，即使分布式锁失效，数据库锁仍能防止同一行被重复处理
• 业务层面：@LockOrder 单据级锁 + 状态检查（如 isOutboundCancel）提供额外保护
• 升级方案：使用Redisson的RedLock算法（多节点），或者用数据库乐观锁（version字段）替代Redis锁

---

Q10：系统中有VisiableThreadPoolTaskExecutor监控类，通过重写execute/submit打印线程池状态。这种方式有什么问题？生产环境你会怎么做？

考察点：线程池监控、可观测性设计

参考回答：

当前方案的优缺点：

优点：零侵入，通过继承ThreadPoolTaskExecutor重写方法，不需要修改业务代码。

缺点：

1. 每次提交任务都打印日志，高并发下日志量巨大，影响性能和磁盘IO
2. 只能看到提交时的快照，看不到历史趋势和峰值
3. 日志分散在各处，不便于集中监控和告警
4. 无法感知线程池的"健康度"——比如队列使用率80%时应该告警，但日志方式做不到

生产环境改进方案：

1. 接入Micrometer + Prometheus：将线程池核心指标（activeCount、queueSize、completedTaskCount、poolSize）暴露为metrics，Grafana可视化
2. 定时采集替代每次打印：用ScheduledExecutorService每10秒采集一次线程池状态，而不是每次提交都打印
3. 告警机制：队列使用率超过80%、活跃线程数接近最大值时触发告警
4. Spring Boot Actuator：自定义Endpoint暴露所有线程池状态，支持HTTP查询

@Bean
public MeterBinder threadPoolMetrics(ThreadPoolTaskExecutor executor) {
    return registry -> {
        Gauge.builder("thread.pool.active", executor, ThreadPoolTaskExecutor::getActiveCount)
            .tag("pool", "inventory").register(registry);
        Gauge.builder("thread.pool.queue.size", executor,
            e -> e.getThreadPoolExecutor().getQueue().size())
            .tag("pool", "inventory").register(registry);
    };
}

---

成果与数据

• 万邑通标签请求通过分治+多线程并发，将原本串行的N次API调用改为并行，请求耗时从线性增长优化为接近单次请求时间
• eBay可售数量上报通过CompletableFuture + 50线程并发 + 4个/批分区 + 三层递进异常处理，单店铺千级商品上报耗时控制在秒级，上报成功率最大化
• 派单系统通过线程池隔离（舱壁模式），万邑通慢请求不再阻塞其他类型派单，整体派单吞吐量提升显著
• 四层锁机制（Redisson分布式锁 + Seata全局锁 + FOR UPDATE行锁 + @LockOrder单据锁）保证了分布式环境下的并发安全，零重复派单事故

---

技术亮点提炼

亮点一：数组按索引写入的无锁保序设计

万邑通标签请求拆分后需要按顺序合并PDF。用 PrintSingleItemLabelResponse[] 数组作为结果容器，每个线程写入自己的索引位置，天然无竞争无需同步，同时保证了合并顺序。配合 CountDownLatch 等待所有线程完成，是"分治-并发-合并"模式的经典实现。

亮点二：三层递进异常处理最大化上报成功率

eBay上报采用"批量4个→逐个重试→记录失败"的递进策略。正常路径API调用次数是N/4，只有失败时才降级为逐个调用。批量失败可能只是1个item有问题，逐个重试精确定位失败item，其他3个仍然成功。

亮点三：线程池隔离的舱壁模式

派单系统为万邑通设计独立的小线程池（2-4线程），与通用派单线程池（5-96线程）隔离。万邑通API响应慢不会占满共享线程池导致其他类型派单饿死。jobTaskExecutor使用AbortPolicy而非CallerRunsPolicy，防止定时调度线程被阻塞。

亮点四：四层锁的精细化并发控制

从集群级（Redisson分布式锁）到事务级（@GlobalLock）到行级（FOR UPDATE）到单据级（@LockOrder），四层锁逐步细化粒度，每层解决不同维度的并发问题，在保证安全的同时最小化锁的范围。

---

简历提炼版

多线程与JUC并发编程实战（万邑通标签 / eBay上报 / 出库派单）

• 设计万邑通标签请求的分治+多线程并发方案，采用 CountDownLatch + 数组按索引保序 + SynchronousQueue线程池（CallerRunsPolicy背压），将大请求拆分为多个子请求并行调用API后合并PDF
• 实现 eBay 可售数量的CompletableFuture多线程上报，50线程并发 + ConcurrentLinkedQueue无锁收集 + "批量4个→逐个重试→记录失败"三层递进异常处理，最大化上报成功率
• 设计出库派单的线程池隔离（舱壁模式）+ 四层锁方案，万邑通慢请求独立线程池隔离，Redisson分布式锁 + Seata全局锁 + FOR UPDATE行锁 + @LockOrder单据锁保证分布式环境并发安全