JAVA异步编排CompletableFuture难以调试的三种可观测方案

大家好，今天我们来聊聊Java异步编程中一个非常重要的工具——CompletableFuture，以及在使用它进行复杂异步编排时，如何解决调试困难的问题。

CompletableFuture 是 Java 8 引入的用于异步编程的强大类。它允许我们以非阻塞的方式执行任务，并将任务的结果传递给后续的处理步骤。通过组合多个 CompletableFuture，我们可以构建复杂的异步流程。然而，随着异步流程的复杂性增加，调试也变得越来越困难。传统的断点调试在异步场景下往往显得力不从心，因为代码的执行顺序不再是线性的，线程切换频繁，难以跟踪。

那么，如何提高 CompletableFuture 编排的可观测性，从而有效地进行调试呢？我将从三个方面详细介绍：日志增强、链路追踪以及指标监控。

一、日志增强：让每个异步步骤都留下痕迹

日志是最基础但也是最有效的可观测性手段。在异步流程中，我们需要确保关键步骤都有日志记录，以便在出现问题时能够快速定位。

1. 简单的日志记录

最简单的做法就是在每个 CompletableFuture 的 thenApply、thenAccept、thenRun 等方法中添加日志。

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class CompletableFutureLoggingExample {

    private static final Logger logger = LoggerFactory.getLogger(CompletableFutureLoggingExample.class);

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            logger.info("Step 1: Generating data");
            return "Hello, World!";
        }).thenApply(data -> {
            logger.info("Step 2: Transforming data. Input: {}", data);
            String transformedData = data.toUpperCase();
            logger.info("Step 2: Transformed data. Output: {}", transformedData);
            return transformedData;
        }).thenAccept(data -> {
            logger.info("Step 3: Processing data. Input: {}", data);
            System.out.println("Processed data: " + data);
        }).exceptionally(ex -> {
            logger.error("Exception occurred: ", ex);
            return null; // Or handle the exception appropriately
        });

        future.get(); // Wait for the completion
    }
}

这段代码在每个 CompletableFuture 的步骤中都添加了日志记录。我们记录了输入和输出数据，以及可能发生的异常。

优点：

简单易用，容易上手。

缺点：

代码冗余，每个步骤都需要手动添加日志。
缺乏上下文信息，难以追踪完整的请求链路。

2. 使用 AOP 统一添加日志

为了解决代码冗余的问题，我们可以使用 AOP (面向切面编程) 来统一添加日志。

import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.concurrent.CompletableFuture;

@Aspect
@Component
public class CompletableFutureLoggingAspect {

    private static final Logger logger = LoggerFactory.getLogger(CompletableFutureLoggingAspect.class);

    @Around("execution(* java.util.concurrent.CompletableFuture.*(..))")
    public Object logCompletableFuture(ProceedingJoinPoint joinPoint) throws Throwable {
        String methodName = joinPoint.getSignature().getName();
        Object[] args = joinPoint.getArgs();

        logger.info("Before executing CompletableFuture method: {}, arguments: {}", methodName, Arrays.toString(args));

        Object result;
        try {
            result = joinPoint.proceed();
            logger.info("After executing CompletableFuture method: {}, result: {}", methodName, result);
        } catch (Throwable throwable) {
            logger.error("Exception occurred while executing CompletableFuture method: {}, arguments: {}", methodName, Arrays.toString(args), throwable);
            throw throwable;
        }

        return result;
    }
}

这段代码使用 Spring AOP 拦截了所有 CompletableFuture 的方法调用，并在方法执行前后记录日志。

优点：

减少了代码冗余，实现了日志的统一管理。
可以记录方法名、参数和返回值，提供更丰富的信息。

缺点：

需要引入 AOP 框架，增加了项目的复杂度。
仍然缺乏上下文信息，难以追踪完整的请求链路。
会影响所有CompletableFuture的操作，可能产生过多的日志。

3. 使用 MDC (Mapped Diagnostic Context) 传递上下文信息

为了追踪完整的请求链路，我们可以使用 MDC 来传递上下文信息。MDC 是一种线程安全的上下文，可以存储与当前线程相关的诊断信息，例如请求 ID、用户 ID 等。

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;

import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

public class CompletableFutureMDCExample {

    private static final Logger logger = LoggerFactory.getLogger(CompletableFutureMDCExample.class);

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        String requestId = UUID.randomUUID().toString();
        MDC.put("requestId", requestId);

        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            logger.info("Step 1: Generating data");
            return "Hello, World!";
        }).thenApply(data -> {
            logger.info("Step 2: Transforming data. Input: {}", data);
            String transformedData = data.toUpperCase();
            logger.info("Step 2: Transformed data. Output: {}", transformedData);
            return transformedData;
        }).thenAccept(data -> {
            logger.info("Step 3: Processing data. Input: {}", data);
            System.out.println("Processed data: " + data);
        }).exceptionally(ex -> {
            logger.error("Exception occurred: ", ex);
            return null; // Or handle the exception appropriately
        });

        future.get();

        MDC.remove("requestId"); // Clean up MDC after the request is finished
    }
}

在这段代码中，我们首先生成一个唯一的请求 ID，并将其放入 MDC 中。然后，在每个 CompletableFuture 的步骤中，日志记录器会自动将请求 ID 包含在日志消息中。这样，我们就可以通过请求 ID 将所有相关的日志消息关联起来，从而追踪完整的请求链路。

优点：

可以传递上下文信息，追踪完整的请求链路。
对代码的侵入性较小，只需要在入口处设置 MDC，并在出口处清除 MDC。

缺点：

需要在每个线程中手动设置和清除 MDC，容易出错。
只适用于单线程上下文，如果涉及到线程池，需要特殊处理。

最佳实践：

使用框架提供的 MDC 集成，例如 Spring Cloud Sleuth。
在线程池中执行任务时，需要将 MDC 从父线程复制到子线程。

二、链路追踪：可视化异步流程的执行路径

链路追踪是一种更加高级的可观测性手段，它可以可视化异步流程的执行路径，帮助我们快速定位瓶颈和错误。

1. 使用 OpenTelemetry 进行链路追踪

OpenTelemetry 是一种开源的可观测性框架，它提供了一套标准的 API 和 SDK，用于收集和导出链路追踪数据。

import io.opentelemetry.api.GlobalOpenTelemetry;
import io.opentelemetry.api.trace.Span;
import io.opentelemetry.api.trace.Tracer;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

public class CompletableFutureOpenTelemetryExample {

    private static final Tracer tracer = GlobalOpenTelemetry.getTracer("CompletableFutureOpenTelemetryExample", "1.0.0");

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Span rootSpan = tracer.spanBuilder("Root Span").startSpan();

        try (var scope = rootSpan.makeCurrent()) {
            CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
                Span span = tracer.spanBuilder("Step 1: Generate data").startSpan();
                try (var s = span.makeCurrent()) {
                    return "Hello, World!";
                } finally {
                    span.end();
                }
            }).thenApply(data -> {
                Span span = tracer.spanBuilder("Step 2: Transform data").startSpan();
                try (var s = span.makeCurrent()) {
                    String transformedData = data.toUpperCase();
                    span.setAttribute("input", data);
                    span.setAttribute("output", transformedData);
                    return transformedData;
                } finally {
                    span.end();
                }
            }).thenAccept(data -> {
                Span span = tracer.spanBuilder("Step 3: Process data").startSpan();
                try (var s = span.makeCurrent()) {
                    System.out.println("Processed data: " + data);
                } finally {
                    span.end();
                }
                span.end();
            }).exceptionally(ex -> {
                rootSpan.recordException(ex);
                return null; // Or handle the exception appropriately
            });

            future.get();
        } finally {
            rootSpan.end();
        }
    }
}

这段代码使用 OpenTelemetry API 创建了 Span，并在每个 CompletableFuture 的步骤中启动和结束 Span。我们还添加了 Attribute 来记录输入和输出数据。

优点：

可以可视化异步流程的执行路径，快速定位瓶颈和错误。
可以记录 Span 的持续时间，分析性能问题。
可以添加 Attribute 和 Event，提供更丰富的上下文信息。

缺点：

需要引入 OpenTelemetry 框架，增加了项目的复杂度。
需要在每个步骤中手动创建和管理 Span，代码侵入性较高。
需要配置 OpenTelemetry Collector 和后端存储，例如 Jaeger 或 Zipkin。

2. 使用 Spring Cloud Sleuth 进行链路追踪

Spring Cloud Sleuth 是 Spring Cloud 提供的链路追踪解决方案，它可以与 Spring Cloud 生态系统无缝集成。

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.sleuth.Tracer;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;

@Service
public class CompletableFutureSleuthExample {

    private static final Logger logger = LoggerFactory.getLogger(CompletableFutureSleuthExample.class);

    @Autowired
    private Tracer tracer;

    public CompletableFuture<String> processData(String input) {
        return CompletableFuture.supplyAsync(() -> {
            logger.info("Step 1: Generating data");
            return "Hello, World!";
        }).thenApply(data -> {
            logger.info("Step 2: Transforming data. Input: {}", data);
            String transformedData = data.toUpperCase();
            logger.info("Step 2: Transformed data. Output: {}", transformedData);
            return transformedData;
        }).thenAccept(data -> {
            logger.info("Step 3: Processing data. Input: {}", data);
            System.out.println("Processed data: " + data);
        }).exceptionally(ex -> {
            logger.error("Exception occurred: ", ex);
            return null; // Or handle the exception appropriately
        }).thenApply(result -> {
            tracer.currentSpan().ifPresent(span -> {
                span.tag("result", String.valueOf(result));
            });
            return result;
        });
    }
}

这段代码使用 Spring Cloud Sleuth 自动创建和管理 Span。我们只需要在需要添加额外信息的步骤中，使用 tracer.currentSpan() 获取当前 Span，并添加 Tag。

优点：

与 Spring Cloud 生态系统无缝集成，配置简单。
自动创建和管理 Span，减少了代码侵入性。

缺点：

只适用于 Spring Cloud 项目。
功能相对简单，不如 OpenTelemetry 灵活。

最佳实践：

选择适合自己项目的链路追踪解决方案。
配置链路追踪采样率，避免收集过多的数据。
使用链路追踪可视化工具，例如 Jaeger 或 Zipkin。

三、指标监控：量化异步流程的性能和健康状况

指标监控是一种主动的可观测性手段，它可以量化异步流程的性能和健康状况，帮助我们及时发现和解决问题。

1. 使用 Micrometer 监控 CompletableFuture 的执行时间

Micrometer 是一种与厂商无关的指标收集库，它可以与多种监控系统集成，例如 Prometheus、InfluxDB 等。

import io.micrometer.core.instrument.MeterRegistry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;

@Service
public class CompletableFutureMicrometerExample {

    private static final Logger logger = LoggerFactory.getLogger(CompletableFutureMicrometerExample.class);

    @Autowired
    private MeterRegistry meterRegistry;

    public CompletableFuture<String> processData(String input) {
        long startTime = System.nanoTime();
        return CompletableFuture.supplyAsync(() -> {
            logger.info("Step 1: Generating data");
            return "Hello, World!";
        }).thenApply(data -> {
            logger.info("Step 2: Transforming data. Input: {}", data);
            String transformedData = data.toUpperCase();
            logger.info("Step 2: Transformed data. Output: {}", transformedData);
            return transformedData;
        }).thenAccept(data -> {
            logger.info("Step 3: Processing data. Input: {}", data);
            System.out.println("Processed data: " + data);
        }).exceptionally(ex -> {
            logger.error("Exception occurred: ", ex);
            return null; // Or handle the exception appropriately
        }).thenApply(result -> {
            long endTime = System.nanoTime();
            meterRegistry.timer("completablefuture.process.time").record(endTime - startTime, java.util.concurrent.TimeUnit.NANOSECONDS);
            return result;
        });
    }
}

这段代码使用 Micrometer 的 Timer 来记录 CompletableFuture 的执行时间。我们首先在方法开始时记录开始时间，然后在方法结束时记录结束时间，并将时间差记录到 Timer 中。

优点：

可以量化异步流程的性能，例如平均执行时间、最大执行时间、吞吐量等。
可以设置告警规则，及时发现性能问题。

缺点：

需要引入 Micrometer 框架，增加了项目的复杂度。
需要在代码中手动记录指标，代码侵入性较高。
需要配置监控系统，例如 Prometheus 或 InfluxDB。

2. 使用 Spring Boot Actuator 监控线程池的指标

Spring Boot Actuator 提供了许多内置的指标，例如 JVM 指标、HTTP 请求指标、线程池指标等。我们可以使用 Actuator 来监控 CompletableFuture 使用的线程池的指标，例如线程池大小、活跃线程数、队列长度等。

优点：

配置简单，无需额外的代码。
提供了丰富的内置指标。

缺点：

只能监控线程池的指标，无法监控单个 CompletableFuture 的指标。
需要 Spring Boot 项目。

最佳实践：

选择适合自己项目的监控系统。
配置合理的告警规则。
定期分析监控数据，优化异步流程的性能。

四、总结与选择建议

可观测方案	优点	缺点	适用场景	复杂度
日志增强	简单易用，容易上手，可以记录关键步骤的信息。	代码冗余，缺乏上下文信息，难以追踪完整的请求链路。	适用于简单的异步流程，或者作为其他可观测方案的补充。	低
链路追踪	可以可视化异步流程的执行路径，快速定位瓶颈和错误，可以记录 Span 的持续时间，分析性能问题。	需要引入链路追踪框架，增加了项目的复杂度，需要在每个步骤中手动创建和管理 Span，代码侵入性较高，需要配置链路追踪 Collector 和后端存储。	适用于复杂的异步流程，需要追踪请求链路，分析性能问题。	高
指标监控	可以量化异步流程的性能和健康状况，可以设置告警规则，及时发现性能问题。	需要引入指标监控框架，增加了项目的复杂度，需要在代码中手动记录指标，代码侵入性较高，需要配置监控系统。	适用于需要量化异步流程的性能和健康状况，需要设置告警规则的场景。	中

选择建议：

对于简单的异步流程，可以使用日志增强来记录关键步骤的信息。
对于复杂的异步流程，可以使用链路追踪来可视化执行路径，分析性能问题。
对于需要量化异步流程的性能和健康状况的场景，可以使用指标监控。
可以将多种可观测方案结合使用，例如使用日志增强来记录详细信息，使用链路追踪来追踪请求链路，使用指标监控来量化性能。

在实际应用中，我们需要根据项目的具体情况选择合适的可观测方案。没有银弹，只有最适合的方案。重要的是，我们要意识到可观测性的重要性，并在开发过程中注重可观测性的设计。

希望今天的分享对大家有所帮助，谢谢！