CompletableFuture 实战：从 OCR 批量识别到 AI 报告生成的并行优化

技术栈：Spring Boot + JDK 8 + MyBatis
场景：附件批量 OCR 识别、AI 帮扶报告生成

摘要

这两个功能都卡在外部 HTTP 调用上：

• OCR 批量识别：一次提交可能包含多个附件，每个文件都要下载、识别、写历史记录；
• AI 帮扶报告：一次生成要执行 5 个 AI 任务，其中 4 个彼此独立，1 个依赖前置结果。

它们刚好代表两种不同的并行语义：

• OCR 是部分成功：个别文件失败，不应该拖垮整批；
• 报告是整体成功：缺一块内容，就应该失败并回滚。

本文基于项目里的真实代码，讲清楚CompletableFuture在这两个场景里的编排方式、线程池配置，以及超时、异常、事务这三个最容易出问题的边界。

一、背景：两个 IO 密集场景

1.1 OCR 批量识别

表单里的附件字段通常是一组 URL，用逗号分隔。识别一个文件要经历：

下载文件 → 计算 MD5 → 调用 OCR 服务 → 保存识别历史 → 合并识别结果

单个文件依赖外部 OCR 服务，耗时不稳定；文件一多，串行执行就是所有文件耗时之和。项目里实测，批量 OCR 从串行约 10 分钟降到约 3 分钟。

1.2 AI 帮扶报告

一次报告生成包含 5 个 AI 任务：

如果串行执行，RT 是 5 个 AI 调用之和；而其中 4 个本来可以同时进行。

两个场景的共同点：大部分任务彼此独立、瓶颈在 IO、请求线程大部分时间都在等。这正是CompletableFuture最合适的用武之地。

二、OCR 批量识别：supplyAsync + get(timeout)

代码位于BetterOcrIdentifyUtil.processBatchBySingleOcr。

2.1 专用线程池

privatestaticfinalExecutorServiceEXECUTOR=newThreadPoolExecutor(5,10,60L,TimeUnit.SECONDS,newLinkedBlockingQueue<>(200),// 给线程加序号newThreadFactoryBuilder().setNameFormat("OcrFastThread-%d").build(),// 满了就让主线程跑，优雅降级newThreadPoolExecutor.CallerRunsPolicy());

OCR 是重 IO、轻 CPU 的任务，核心线程 5、最大 10 足够；队列 200 用来缓冲突发文件；CallerRunsPolicy保证池满时任务不会丢，而是退化为调用线程执行。

2.2 每个文件一个 future

List<CompletableFuture<OcrResultData<T>>>futures=Arrays.stream(urls).filter(StrUtil::isNotBlank).map(StrUtil::trim).map(url->CompletableFuture.supplyAsync(()->{// 先查缓存OcrIdentifyResponse<OcrResultData<T>>cachedResp=OcrIdentifyCacheUtil.getSingleCache(url);if(cachedResp!=null){returncachedResp.getData();}// 下载 + 计算 MD5 + 记录临时文件DownloadResultdr=downloadAndMd5(url,createDownloadPath());if(dr==null){returnnull;}tempPaths.add(dr.getPath());try{// 调用 OCR 服务 + 保存识别历史OcrIdentifyResponse<OcrResultData<T>>res=sendFileToUrl(url,newFile(dr.getPath()),docType);if(res!=null&&res.getCode()==200&&res.getData()!=null){getHistoryService().mergeInto(newOcrHistory(Md5Util.urlKeyMd5(url),res.getData().getOriginalJson()));returnres.getData();}}catch(Exceptione){log.error("并发识别单任务异常: {}",url,e);}returnnull;},EXECUTOR)).collect(Collectors.toList());

这里有四个设计点：

（1）任务粒度是“一个文件”

每个 URL 对应一个supplyAsync，下载、识别、写历史全部放在同一个任务里。任务之间没有业务状态依赖，唯一的共享是线程安全的tempPaths，失败也容易隔离。

（2）缓存命中直接返回

识别前先查单文件缓存。批量场景里经常出现重复提交，缓存命中时可以完全跳过下载和 OCR，这是比并行更彻底的优化。

（3）任务内部捕获异常，失败返回null

}catch(Exceptione){log.error("并发识别单任务异常: {}",url,e);}returnnull;

OCR 批量识别的产品语义是“能识别多少算多少”。如果让异常从任务里抛出去，后续get()会抛ExecutionException，整批结果都拿不到。

（4）临时文件列表必须线程安全

List<String>tempPaths=newCopyOnWriteArrayList<>();

多个下载线程会同时add文件路径，普通ArrayList在这里会出现并发写问题。

2.3 结果收集：为什么用 get(timeout) 而不是 allOf

List<OcrResultData<T>>successList=futures.stream().map(f->{try{// 这里的超时时间应略大于 OkHttp 的 readTimeoutreturnf.get(13,TimeUnit.MINUTES);}catch(Exceptione){// 物理中断f.cancel(true);log.warn("OCR TASK TIMEOUT");returnnull;}}).filter(Objects::nonNull).collect(Collectors.toList());OcrIdentifyResponse<OcrResultData<T>>finalResp=buildResponse(200,"success");finalResp.setData(OcrResultMergeUtil.mergeResults(successList,DocTypeEnum.getClazzByCode(docType)));returnfinalResp;

这里没有用allOf(...).join()，而是逐个get(13, TimeUnit.MINUTES)，原因有两个：

1. 需要单任务超时。allOf.join()没有超时参数，只要有一个任务卡住，调用线程就会一直等；
2. 需要吞掉单任务失败。get()抛异常时返回null，后续过滤掉即可，不会影响其他文件。

allOf适合“全部成功才有意义”的场景；OCR 这种“允许部分失败”的场景，逐个get更合适。

2.4 两层超时是配套的

外层get等 13 分钟，底层 OkHttp 的配置是：

newOkHttpClient.Builder().connectTimeout(6,TimeUnit.MINUTES).readTimeout(12,TimeUnit.MINUTES).writeTimeout(2,TimeUnit.MINUTES).build();

读超时 12 分钟，外层等待 13 分钟。外层比内层多 1 分钟，保证常规的读取超时由 OkHttp 先返回。不过connectTimeout还有 6 分钟，最坏情况下“连接 6 分钟 + 读取 12 分钟”会超过 13 分钟，这时外层会先放弃。严格来说这是一个理论上的缺口，压测时可以关注连接阶段的耗时。

2.5 cancel(true) 不是强制终止

f.cancel(true);

cancel(true)只是向执行线程发送中断信号。任务是否响应中断，取决于它有没有处在可中断的阻塞点上。OkHttp 的同步execute()等待 socket 读时，不一定能立刻响应中断。

所以真正兜底的是两层超时：

1. OkHttp 的readTimeout限制单次调用；
2. get(13, MINUTES)限制请求线程的等待时间。

cancel(true)是“尽力而为”的清理，不能当成强制的任务终止。

2.6 临时文件清理

方法结束时要删除下载的临时文件：

try{returnprocessBatchBySingleOcr(fileUrls,docType,tempPaths);}finally{asyncCleanFiles(tempPaths);}

清理本身也提交到同一个线程池异步执行，避免删除文件拖慢响应。但要注意：清理任务和 OCR 任务共用一个池，文件量特别大时可能互相排队。如果清理成为瓶颈，可以单独给一个小池，或者改用延迟队列。

2.7 从“下载并行”到“全链路并行”

项目里还有一版OcrIdentifyUtil，它只把下载并行化，识别仍然是一次批量调用：

List<CompletableFuture<Map.Entry<String,String>>>futures=Arrays.stream(urls).filter(StrUtil::isNotBlank).map(url->CompletableFuture.supplyAsync(()->{Stringpath=null;try{path=downloadSingleFile(url.trim(),baseDownloadPath);}catch(Exceptione){log.error("下载异常: {}",url,e);}return(Map.Entry<String,String>)newAbstractMap.SimpleEntry<>(url,path);},EXECUTOR)).collect(Collectors.toList());CompletableFuture.allOf(futures.toArray(newCompletableFuture[0])).join();

两版的关键差异：

从“下载并行”到“全链路并行”，是这类 IO 密集任务最常见的演进路径。并行范围越靠近真正的瓶颈，收益越明显。

三、AI 帮扶报告：supplyAsync + thenApplyAsync

代码位于HelpReportServiceImpl.gen。

3.1 报告任务使用的线程池

报告任务通过@Qualifier("myExecutor")注入 Spring 管理的线程池：

@Bean("myExecutor")publicExecutormyExecutor(){ThreadPoolTaskExecutorexecutor=newThreadPoolTaskExecutor();executor.setCorePoolSize(10);executor.setMaxPoolSize(20);executor.setQueueCapacity(200);executor.setKeepAliveSeconds(60);executor.setThreadNamePrefix("taskExecutor-");executor.setRejectedExecutionHandler(newThreadPoolExecutor.CallerRunsPolicy());returnexecutor;}

核心 10、最大 20、队列 200。一次报告只涉及 5 个任务，这个容量足够；即使多个报告请求并发，队列也能起到缓冲作用。

3.2 编排代码

// 并行执行：不合格项目详细分析 / 历史抽检风险提示 / 相关案例// 生产工艺参考依赖 analysisDetails 的整改建议列表，故用 thenApplyAsync 串联CompletableFuture<List<HelpReport.AnalysisDetail>>analysisFuture=CompletableFuture.supplyAsync(()->getAnalyzeItem(dto.getFoodName(),dto.getEnterpriseName(),dto.getSampleLinkName(),inspectionItems),myExecutor);CompletableFuture<String>riskSummaryFuture=CompletableFuture.supplyAsync(()->getRiskSummaryStrNew(upFoodInspectionItems),myExecutor);CompletableFuture<String>relatedCaseFuture=CompletableFuture.supplyAsync(()->getRelatedCaseStr(dto,inspectionItems),myExecutor);CompletableFuture<String>upInspectionItemsSummary=CompletableFuture.supplyAsync(()->getupInspectionItemsSummary(dto,upInspectionItems),myExecutor);// 依赖任务：等 analysisFuture 完成后再执行CompletableFuture<String>productionProcessFuture=analysisFuture.thenApplyAsync(analysisDetails->{List<String>suggestions=extractSuggestions(analysisDetails);returngetProductionProcessStr(dto,inspectionItems,suggestions);},myExecutor);// 等待全部完成CompletableFuture.allOf(analysisFuture,riskSummaryFuture,relatedCaseFuture,productionProcessFuture).join();// 取值并落库report.setAnalysisDetails(JSON.toJSONString(analysisFuture.join()));report.setRiskSummary(riskSummaryFuture.join());report.setRelatedCases(relatedCaseFuture.join());report.setProductionProcess(productionProcessFuture.join());report.setUpInspectionItemsSummary(upInspectionItemsSummary.join());helpReportMapper.insert(report);

执行关系：

analysisFuture ──thenApplyAsync──> productionProcessFuture ──┐ riskSummaryFuture ──────────────────────────────────────────┤ relatedCaseFuture ──────────────────────────────────────────┼──> allOf(...).join() productionProcessFuture ────────────────────────────────────┘ upInspectionItemsSummary ──(后面单独 join 兜底)────────────────┘

3.3 用 thenApplyAsync 表达依赖

生产工艺参考不能和不合格项分析同时启动，它需要从analysisDetails里提取整改建议。所以不能用 5 个平级的supplyAsync：

CompletableFuture<String>productionProcessFuture=analysisFuture.thenApplyAsync(analysisDetails->{List<String>suggestions=extractSuggestions(analysisDetails);returngetProductionProcessStr(dto,inspectionItems,suggestions);},myExecutor);

thenApplyAsync做两件事：

1. 把“依赖关系”写进代码结构，而不是靠手工阻塞等待；
2. 显式指定后续任务也使用myExecutor，线程使用可控。

如果写成thenApply，回调可能在上游完成的那个线程里执行，虽然也能跑，但线程归属不明确，日志和排查会更麻烦。

3.4 allOf 只负责等待，不负责收集

allOf的返回值是CompletableFuture<Void>，它只表示“全部完成”，不携带任何结果。所以后面还要逐个join()取值。

这里还有一个细节：代码里的allOf没有包含upInspectionItemsSummary，只是靠后面的join()兜底才没有出问题。

CompletableFuture.allOf(analysisFuture,riskSummaryFuture,relatedCaseFuture,productionProcessFuture).join();

如果后续重构时把单独的那句upInspectionItemsSummary.join()删掉，就会漏掉这块内容。allOf的参数列表最好写全，这样语义更清晰，也不依赖某一次join恰好兜底。

3.5 事务边界：异步线程不在事务里

gen()上有@Transactional(rollbackFor = Exception.class)。

关键事实：事务上下文不会传播到myExecutor的线程里。@Transactional绑定的是当前请求线程，异步任务拿到的是另一个线程，既不在同一个事务里，也无法参与回滚。

这段代码的做法是正确的：

• AI 调用全部在异步线程完成；
• 数据库写入放在主线程allOf.join()之后。

不要在两个异步任务里各写一次库，然后期待它们和主事务一起回滚——它们根本不在同一个事务里。

3.6 失败语义：整体成功，还是允许降级

这几个 AI 方法没有用exceptionally/handle做统一降级。底层 HTTP 工具虽然会吞掉网络异常并返回失败结果（null或code=500），但上层对失败结果的解析并不总是安全的：例如拿到null的data后继续JSON.parseObject并取值，仍可能抛出空指针。

一旦某个任务抛出异常：

1. 对应的 future 异常完成；
2. allOf(...).join()抛出CompletionException；
3. @Transactional生效，报告落库回滚。

如果业务上希望“某个 AI 模块失败也返回报告”，需要显式降级：

CompletableFuture<String>riskSummaryFuture=CompletableFuture.supplyAsync(()->getRiskSummaryStrNew(upFoodInspectionItems),myExecutor).exceptionally(e->{log.error("历史风险生成失败",e);return"[]";});

是否降级不是技术问题，而是产品语义问题：报告缺一块内容，能不能接受。

3.7 补充：任务内部仍是串行

getAnalyzeItem内部会按不合格项逐个调用 AI 接口：

for(HelpReport.InspectionIteminspectionItem:inspectionItems){// 每个不合格项调用一次 AI 接口Stringreport=AiReportUtil.analyzeItemContent(body);// ...}

外层虽然只占一个 future，单任务内部仍是串行。如果不合格项数量很大，可以在方法内部再做一层并行，但同样要受线程池容量约束，不能无限制地拆。

四、两种并行语义的对比

最核心的差异是失败策略：

• OCR 是“能识别多少算多少”，所以任务内吞异常；
• 报告是“缺内容就不可用”，所以让异常冒泡、事务回滚。

先确定失败语义，再选 API，比先写CompletableFuture再想异常怎么处理要清晰得多。

五、API 与线程池速查

5.1 常用 API

5.2 线程池配置

六、八条经验

1. 显式传线程池。不传就用ForkJoinPool.commonPool()，IO 阻塞会占满全局公共池，影响所有依赖它的代码。
2. 线程池必须命名。OcrFastThread-%d、taskExecutor-%d这样的名字，出问题时能直接从线程栈定位。
3. 先确定失败语义。要部分成功就在任务内try/catch；要整体成功就让异常冒泡。不要让allOf.join()的默认行为替你做决定。
4. allOf只负责等待，不负责收集。结果要逐个join，参数列表要写全。
5. 外部 IO 一定要有超时。外层等待时间略大于底层 socket 超时，避免请求线程无限挂起。
6. cancel(true)不等于强制终止。它只是发中断信号，真正兜底的是底层超时和外层等待超时。
7. 注意事务边界。@Transactional不会传播到异步线程，异步任务里的数据库操作是独立事务。
8. 并行度受下游资源约束。OCR 服务、AI 工作流、线程池队列都有自己的容量上限，压测要看池的等待时间，而不只是接口 RT。

七、结语

CompletableFuture本身不难，难的是并行之后带来的边界问题：

• 异常是快速失败还是部分成功；
• 结果由谁收集、顺序是否重要；
• 任务卡住时，谁来超时、谁来取消；
• 事务是否还能保持原来的假设。

回到这两个功能：

• OCR 批量识别要“尽力而为”，所以逐个get(timeout)、失败返回null、过滤后合并；
• 帮扶报告要“整体可用”，所以allOf.join()、异常冒泡、事务回滚。

同样是CompletableFuture，因为业务语义不同，落地方式就完全不同。把失败语义和资源边界想清楚，再写并行代码，才会既快又稳。