基于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 17Spring Boot 3.2.5OpenTelemetry API 1.34.0Lombok 1.18.30核心Maven依赖dependencies dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-web/artifactId /dependency dependency groupIdio.opentelemetry/groupId artifactIdopentelemetry-api/artifactId version1.34.0/version /dependency dependency groupIdio.opentelemetry/groupId artifactIdopentelemetry-sdk/artifactId version1.34.0/version /dependency dependency groupIdio.opentelemetry/groupId artifactIdopentelemetry-exporter-otlp/artifactId version1.34.0/version /dependency dependency groupIdorg.projectlombok/groupId artifactIdlombok/artifactId optionaltrue/optional /dependency /dependencies1. 自定义任务上下文封装任务状态、进度等业务数据用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异步处理时更新进度并埋点异常时更新状态并记录异常Spanimport 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 ConcurrentHashMapString, 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 MapString, 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 filetest.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的跨服务传播能力即可实现全链路的进度追踪与观测。