基于CompletableFuture与OpenTelemetry实现文件上传异步任务进度精准追踪-酒店常州论坛

基于CompletableFuture与OpenTelemetry实现文件上传异步任务进度精准追踪

问题背景

在电商商品批量导入、媒体平台视频转码、企业报表生成等业务场景中，普遍采用「文件上传+异步处理」的交互模式：用户上传文件后服务端立即返回任务ID，后续通过任务ID查询处理进度。传统实现普遍存在两类核心痛点： 1.进度管理不可靠：多数方案用静态Map存储任务状态，无并发控制，多用户并发查询时易出现进度错乱；异步任务异常时状态未及时更新，用户长时间看到「处理中」无反馈。 2.问题排查效率低：微服务架构下上传服务、存储服务、处理服务拆分，异步任务出问题时日志分散在各服务节点，链路断裂，排查需要跨服务翻日志，耗时极长。

若引入分布式任务调度框架（如XXL-Job）又会带来较高的运维成本，对于中小规模（日任务量10万级以下）的场景并不划算。本文提出一种轻量方案，通过CompletableFuture负责业务层面的异步编排与进度计算，OpenTelemetry负责全链路观测，两者配合解决上述痛点。

方案设计

明确两项技术的职责边界，避免强行拼接： -CompletableFuture作为核心进度管理组件：利用JDK原生的异步编排能力实现任务执行、状态流转与进度计算，无需额外引入中间件，通过封装任务上下文管理进度、状态等业务数据，对外提供统一的查询接口。 -OpenTelemetry作为观测层组件：负责串联异步任务的全生命周期链路，从文件上传请求发起、任务提交、处理节点执行到最终完成/异常，全链路打点，将进度、当前步骤、异常上下文等业务数据嵌入链路事件中，出问题时可直接在追踪平台定位根因，无需翻日志。

两者协作关系为：CompletableFuture在任务执行的关键节点（如开始解析文件、处理到指定行数、写入数据库）触发OpenTelemetry的Span事件上报，同时OpenTelemetry将上传请求的父Span与异步任务的子Span关联，保证链路连贯性。

关键原理

1. CompletableFuture进度管理原理

CompletableFuture原生未提供进度查询API，因此需要封装一层AsyncTaskContext作为任务上下文容器，内部用AtomicInteger存储0-100的进度值，用枚举存储任务状态（待处理、处理中、成功、失败），同时存储任务ID、文件信息、异常信息等业务数据。任务执行时通过supplyAsync提交到自定义线程池，利用thenApply、thenAccept等异步回调方法在关键节点更新上下文中的进度与状态。核心注意点：必须自定义线程池，禁止使用CompletableFuture默认的ForkJoinPool，避免异步任务中的阻塞操作占满公共线程池，影响服务其他功能。

2. OpenTelemetry链路串联原理

默认情况下CompletableFuture的异步线程会丢失父线程的OpenTelemetry上下文，导致异步任务Span与上传请求Span断开。因此需要手动传播上下文：上传请求进来时从当前Context中取出任务ID，传递给异步任务；异步任务执行时手动创建子Span，设置父Span为上传请求的Span，同时将Span设置到当前线程的Context中，保证后续事件都属于该Span。关键节点添加的自定义事件会携带进度、处理行数、异常信息等业务属性，直接嵌入链路数据中。

完整示例

环境依赖

JDK 17
Spring Boot 3.2.5
OpenTelemetry API 1.34.0
Lombok 1.18.30

核心Maven依赖：

<dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>io.opentelemetry</groupId> <artifactId>opentelemetry-api</artifactId> <version>1.34.0</version> </dependency> <dependency> <groupId>io.opentelemetry</groupId> <artifactId>opentelemetry-sdk</artifactId> <version>1.34.0</version> </dependency> <dependency> <groupId>io.opentelemetry</groupId> <artifactId>opentelemetry-exporter-otlp</artifactId> <version>1.34.0</version> </dependency> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <optional>true</optional> </dependency> </dependencies>

1. 自定义任务上下文

封装任务状态、进度等业务数据，用AtomicInteger保证进度更新的并发安全：

import io.opentelemetry.api.trace.Span; import lombok.Data; import java.util.Date; @Data public class AsyncTaskContext { private String taskId; private String fileName; private TaskStatus status; // 用AtomicInteger保证进度更新的原子性和可见性 private AtomicInteger progress = new AtomicInteger(0); private String errorMessage; private Date updateTime = new Date(); public void setProgress(int progress) { this.progress.set(progress); this.updateTime = new Date(); } public int getProgress() { return progress.get(); } } // 任务状态枚举 public enum TaskStatus { PENDING, PROCESSING, SUCCESS, FAILED }

2. OpenTelemetry配置类

初始化Tracer，配置控制台导出器方便本地测试，生产环境可替换为Jaeger、Prometheus等后端：

import io.opentelemetry.api.OpenTelemetry; import io.opentelemetry.api.common.Attributes; import io.opentelemetry.api.trace.Tracer; import io.opentelemetry.context.propagation.ContextPropagators; import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporter; import io.opentelemetry.sdk.OpenTelemetrySdk; import io.opentelemetry.sdk.resources.Resource; import io.opentelemetry.sdk.trace.SdkTracerProvider; import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor; import io.opentelemetry.semconv.resource.attributes.ResourceAttributes; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class OpenTelemetryConfig { @Bean public Tracer tracer() { // 配置链路导出器，这里用控制台输出，生产可替换为Jaeger导出器 OtlpGrpcSpanExporter spanExporter = OtlpGrpcSpanExporter.builder() .setEndpoint("http://localhost:4317") .build(); SdkTracerProvider tracerProvider = SdkTracerProvider.builder() .addSpanProcessor(SimpleSpanProcessor.create(spanExporter)) .setResource(Resource.getDefault() .merge(Resource.create(Attributes.of(ResourceAttributes.SERVICE_NAME, "file-upload-service")))) .build(); OpenTelemetrySdk openTelemetry = OpenTelemetrySdk.builder() .setTracerProvider(tracerProvider) .setPropagators(ContextPropagators.create(io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator.getInstance())) .build(); // 设置全局OpenTelemetry实例 OpenTelemetry.setGlobalOpenTelemetry(openTelemetry); return openTelemetry.getTracer("file-upload-processor"); } }

3. 异步任务处理服务

核心逻辑：提交任务时创建上下文存入ConcurrentHashMap，异步处理时更新进度并埋点，异常时更新状态并记录异常Span：

import io.opentelemetry.api.trace.Span; import io.opentelemetry.api.trace.Tracer; import io.opentelemetry.api.common.AttributeKey; import io.opentelemetry.api.common.Attributes; import io.opentelemetry.context.Scope; import org.springframework.stereotype.Service; import org.springframework.web.multipart.MultipartFile; import java.util.Map; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.atomic.AtomicInteger; @Service public class AsyncFileProcessService { // 生产环境可替换为Redis存储任务上下文，避免服务重启丢状态 private final ConcurrentHashMap<String, AsyncTaskContext> taskContextMap = new ConcurrentHashMap<>(); private final Tracer tracer; // 自定义异步任务线程池，禁止使用默认ForkJoinPool private final ThreadPoolExecutor asyncTaskExecutor = new ThreadPoolExecutor( 4, 8, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), new ThreadFactory() { private final AtomicInteger threadNum = new AtomicInteger(0); @Override public Thread newThread(Runnable r) { Thread t = new Thread(r, "async-file-process-" + threadNum.incrementAndGet()); t.setDaemon(true); return t; } }, new ThreadPoolExecutor.CallerRunsPolicy() ); public AsyncFileProcessService(Tracer tracer) { this.tracer = tracer; } /** * 提交异步任务 */ public String submitTask(MultipartFile file) { String taskId = UUID.randomUUID().toString().replace("-", ""); AsyncTaskContext context = new AsyncTaskContext(); context.setTaskId(taskId); context.setFileName(file.getOriginalFilename()); context.setStatus(TaskStatus.PROCESSING); taskContextMap.put(taskId, context); // 提交异步任务到自定义线程池 CompletableFuture.supplyAsync(() -> processFile(file, context), asyncTaskExecutor) .exceptionally(ex -> { context.setStatus(TaskStatus.FAILED); context.setErrorMessage(ex.getMessage()); context.setProgress(100); // 记录异常Span事件 Span errorSpan = tracer.spanBuilder("async-task-exception") .setParent(Span.current().getSpanContext()) .startSpan(); errorSpan.recordException(ex); errorSpan.addEvent("task_failed", Attributes.of(AttributeKey.stringKey("task_id"), taskId)); errorSpan.end(); return null; }); return taskId; } /** * 模拟文件处理逻辑 */ private String processFile(MultipartFile file, AsyncTaskContext context) { // 创建异步任务子Span，关联上传请求的父Span Span processSpan = tracer.spanBuilder("async-file-process") .setParent(Span.current().getSpanContext()) .startSpan(); // 将Span设置到当前线程Context，保证后续事件属于该Span try (Scope ignored = processSpan.makeCurrent()) { processSpan.addEvent("task_start", Attributes.of( AttributeKey.stringKey("task_id"), context.getTaskId(), AttributeKey.stringKey("file_name"), context.getFileName() )); // 模拟处理10000行Excel数据 int totalRows = 10000; int batchSize = 1000; for (int i = 0; i < totalRows; i += batchSize) { // 模拟IO处理耗时 Thread.sleep(500); int progress = (int) ((i + batchSize) * 100.0 / totalRows); context.setProgress(Math.min(progress, 100)); // 添加进度事件到链路 processSpan.addEvent("process_progress", Attributes.of( AttributeKey.stringKey("task_id"), context.getTaskId(), AttributeKey.longKey("processed_rows"), i + batchSize, AttributeKey.longKey("total_rows"), totalRows) ); } context.setProgress(100); context.setStatus(TaskStatus.SUCCESS); processSpan.addEvent("task_success", Attributes.of(AttributeKey.stringKey("task_id"), context.getTaskId())); return "success"; } catch (InterruptedException e) { Thread.currentThread().interrupt(); context.setStatus(TaskStatus.FAILED); context.setErrorMessage("任务被中断"); throw new RuntimeException(e); } finally { processSpan.end(); } } public AsyncTaskContext getTaskContext(String taskId) { return taskContextMap.get(taskId); } }

4. 上传与进度查询接口

import org.springframework.web.bind.annotation.*; import org.springframework.web.multipart.MultipartFile; import java.util.Map; @RestController @RequestMapping("/api/file") public class FileUploadController { private final AsyncFileProcessService asyncFileProcessService; public FileUploadController(AsyncFileProcessService asyncFileProcessService) { this.asyncFileProcessService = asyncFileProcessService; } @PostMapping("/upload") public Map<String, String> upload(@RequestParam("file") MultipartFile file) { String taskId = asyncFileProcessService.submitTask(file); return Map.of("taskId", taskId, "message", "文件上传成功，正在异步处理中"); } @GetMapping("/progress/{taskId}") public AsyncTaskContext getProgress(@PathVariable String taskId) { AsyncTaskContext context = asyncFileProcessService.getTaskContext(taskId); if (context == null) { throw new RuntimeException("任务不存在"); } return context; } }

测试方式

启动服务后上传文件：

curl -F "file=@test.xlsx" http://localhost:8080/api/file/upload

返回示例：

{ "taskId": "a1b2c3d4e5f6", "message": "文件上传成功，正在异步处理中" }

通过返回的taskId查询进度：

curl http://localhost:8080/api/file/progress/a1b2c3d4e5f6

返回示例：

{ "taskId": "a1b2c3d4e5f6", "fileName": "test.xlsx", "status": "PROCESSING", "progress": 50, "errorMessage": null, "updateTime": "2024-05-20T14:30:00" }

同时可在Jaeger或控制台看到完整的链路数据，上传请求Span与异步处理Span关联，包含各个进度节点的属性信息。

常见问题

为什么必须自定义线程池？CompletableFuture无参异步方法默认使用公共ForkJoinPool，若异步任务包含阻塞操作（如IO、数据库查询），会占满公共线程池，导致Stream并行流、其他异步任务卡住，因此必须为异步任务创建独立线程池，配置合理的核心线程数、队列大小和拒绝策略。
异步任务Span与上传请求Span关联不上怎么办？异步线程默认不会继承父线程的OpenTelemetry上下文，需要在异步任务执行时手动将父Span的Context设置为当前Context，示例中通过processSpan.makeCurrent()实现，跨服务场景下需要用W3CTraceContextPropagator将上下文传递到下游服务。
进度查询出现数据不一致？进度更新和查询是并发执行的，必须用AtomicInteger或volatile修饰进度字段，保证内存可见性，避免出现进度回退或查询到旧值的问题。

适用边界与关键取舍

适用边界

该方案适合单服务内或微服务架构下，日任务量10万级以下、单任务执行时间在几分钟到2小时之间的异步任务场景，如文件导入导出、媒体转码、数据同步等。若为日任务量百万级以上的大规模分布式任务场景，CompletableFuture的单机内存存储方案不可靠，需结合XXL-Job等分布式任务调度框架管理任务状态，OpenTelemetry仍可用于跨服务链路追踪。

关键取舍

任务状态存储选内存还是持久化：内存存储性能高、实现简单，但服务重启会丢失进行中的任务状态，适合对进度实时性要求高、任务执行时间短的场景；持久化存储（Redis/MySQL）可靠性高，但存在IO开销，进度更新有延迟，适合任务执行时间长、不能丢状态的场景。
链路采样率选择：全量采集可保证所有问题可追溯，但存储和计算成本高，适合小规模场景；大规模场景下可配置采样率，比如只采集异常链路、10%的正常链路，平衡观测成本和排查效率。

易踩坑细节

CompletableFuture的complete方法仅能生效一次，后续调用无效，因此更新进度时不要用complete方法，而是通过修改上下文中的AtomicInteger更新进度，避免进度更新失败。另外自定义线程池不要和业务其他线程池混用，避免资源争抢。

总结

本方案通过职责分离的设计，让CompletableFuture专注业务层面的异步编排与进度计算，OpenTelemetry专注非功能层面的链路观测，两者配合既解决了传统方案进度不准的问题，又解决了异步任务排查困难的问题。相比引入额外的任务调度中间件，该方案实现轻量，适合中小规模文件上传异步处理场景；若需扩展到分布式场景，只需将任务状态持久化，结合OpenTelemetry的跨服务传播能力，即可实现全链路的进度追踪与观测。

企业官网建设流程全解析