Java中的异步操作CompletableFuture示例详解_Java

一、completablefuture概述

completablefuture 是 java 8 引入的一个功能强大的异步编程工具，它实现了future和completionstage接口，提供了更灵活、更强大的异步任务编排能力。相比传统的future，它支持非阻塞的链式调用和组合操作。

二、completablefuture的核心特性

异步任务的创建与执行

异步任务是completablefuture的基础单元。关键在于区分任务的发起和结果的获取是两个独立的时机，这正是异步的本质。

//最简单的异步调用，默认使用 forkjoinpool.commonpool()
completablefuture<integer> computefuture = completablefuture.supplyasync(() -> {
    thread.sleep(1000);  // 模拟耗时计算
    return 42;
});

completablefuture底层是使用线程池帮我们去完成异步的一个调用，所以，我们也可以自定义线程池来实现异步操作。

//自定义线程池
executorservice bizexecutor = executors.newfixedthreadpool(5, r -> {
    thread t = new thread(r);
    t.setname("bizexecutor-" + t.getid());
    return t;
});

实际开发中，强烈建议始终指定自定义线程池，避免业务代码占用公共线程池影响系统稳定性。

任务结果的链式处理

这是completablefuture最优雅的特性，它将回调地狱转换为流畅的管道。其中有三个重要的方法

thenapply：接受输入，产生输出，实际用作转换函数
thenaccept：接受输入，无输出，作为数据的消费者
thenrun：无输入，无输出，可用作通知等操作

// 电商订单处理流水线示例
completablefuture<order> orderfuture = completablefuture
    // 阶段1：创建订单（有返回值）
    .supplyasync(() -> orderservice.createorder(request), orderexecutor)
    // 阶段2：验证库存（转换结果）
    .thenapplyasync(order -> {
        inventoryservice.checkstock(order.getitems());
        return order.markasvalidated();
    }, inventoryexecutor)
    // 阶段3：扣减库存（消费结果，无返回）
    .thenacceptasync(order -> {
        inventoryservice.deductstock(order.getitems());
        logger.info("库存扣减完成，订单号：{}", order.getid());
    }, inventoryexecutor)
    // 阶段4：发送通知（纯副作用操作）
    .thenrunasync(() -> {
        notificationservice.sendordercreated();
    }, notificationexecutor);

带有async后缀的方法(如thenapplyasync)会在新线程中执行，而不带后缀的会在前一个任务完成的线程中执行。在io密集型场景使用async版本可提高并发度。

多个任务组合

现实业务中很少有单一异步任务，更多是多个任务的协同。completablefuture提供了丰富的组合模式。

thencombine：等待两个都完成，然后合并
thencompose：前一个完成后再开始下一个
allof：等待所有（3+个）任务完成

// 场景：用户详情页需要聚合多个服务的数据
public completablefuture<userprofile> getuserprofile(string userid) {
    // 并行调用三个独立服务
    completablefuture<userinfo> userinfofuture = 
        userservice.getuserinfoasync(userid);
    completablefuture<list<order>> ordersfuture = 
        orderservice.getuserordersasync(userid);
    completablefuture<list<address>> addressesfuture = 
        addressservice.getuseraddressesasync(userid);
    // thencombine等待两个都完成，然后合并
    completablefuture<userwithorders> userwithorders = userinfofuture
        .thencombine(ordersfuture, (user, orders) -> {
            return new userwithorders(user, orders);
        });
    // thencompose前一个完成后再开始下一个
    completablefuture<string> personalizedgreeting = userinfofuture
        .thencompose(user -> 
            greetingservice.getgreetingasync(user.getlanguage(), user.getname())
        );
    // allof 等待所有（3+个）任务完成
    return completablefuture.allof(
            userwithorders, 
            addressesfuture, 
            personalizedgreeting
        )
        .thenapply(v -> {
            // 所有都完成后的聚合
            return new userprofile(
                userwithorders.join().getuser(),
                userwithorders.join().getorders(),
                addressesfuture.join(),
                personalizedgreeting.join()
            );
        });
}

这里有个要注意的点：allof本身返回的是 completablefuture<void>，不包含各子任务的结果。需要像上面示例中通过 join()或 get()来获取，但注意这不会导致死锁，因为此时所有子任务已确定完成。

异常处理

异步中的异常处理比同步更复杂，因为异常和业务代码在时间、空间上都是解耦的。completablefuture 的异常处理是功能型、非侵入式的。

// 健壮的数据处理管道
completablefuture<report> reportfuture = completablefuture
    .supplyasync(() -> datafetcher.fetchdata(), dataexecutor)
    // exceptionally异常时提供降级值
    .exceptionally(ex -> {
        log.warn("数据获取失败，使用缓存", ex);
        return cacheservice.getcacheddata();
    })
    // 对数据做处理，同时捕获处理中的异常
    .thenapplyasync(data -> {
        try {
            return dataprocessor.process(data);
        } catch (processingexception e) {
            throw new completionexception("处理失败", e);
        }
    }, processexecutor)
    // handle统一处理成功和失败
    .handle((processeddata, ex) -> {
        if (ex != null) {
            // 失败路径
            metrics.recordfailure(ex);
            return report.errorreport("系统繁忙，请稍后重试");
        } else {
            // 成功路径
            return report.successreport(processeddata);
        }
    })
    // whencomplete无论成功失败都执行（类似finally）
    .whencomplete((report, ex) -> {
        // 资源清理、日志记录等
        cleanresources();
        log.info("报告生成完成，状态：{}", 
            ex == null ? "成功" : "失败");
    });

exceptionally和handle 的区别：

exceptionally：只在异常时触发，必须返回一个恢复值
handle：无论成功失败都触发，通过第二个参数判断状态

建议在异步链的末尾使用 handle或 whencomplete做统一的最终处理，而不是在每个阶段都捕获异常。

三、性能优化与注意事项

避免阻塞主线程

在web请求线程中调用get()会阻塞整个请求

错误写法：

public result handlerequest() {
    completablefuture<data> future = fetchdataasync();
    return future.get();  // 阻塞，浪费容器线程
}

接下来展示正确写法：

//完全异步
public completablefuture<result> handlerequestasync() {
    return fetchdataasync()
        .thenapply(data -> converttoresult(data))
        .exceptionally(ex -> result.error("处理失败"));
}

合理使用线程池资源

线程池隔离原则：不同业务、不同重要级别的任务使用不同的线程池，避免相互影响。

根据任务类型的不同配置不同的线程池参数

public class threadpoolconfig {
    // cpu密集型：核心数+1
    @bean("cpuintensivepool")
    public executorservice cpuintensivepool() {
        int coresize = runtime.getruntime().availableprocessors() + 1;
        return new threadpoolexecutor(
            coresize, coresize * 2, 60l, timeunit.seconds,
            new linkedblockingqueue<>(1000),
            new namedthreadfactory("cpu-pool")
        );
    }
    // io密集型：可设置较大线程数
    @bean("iointensivepool")  
    public executorservice iointensivepool() {
        return new threadpoolexecutor(
            20, 100, 60l, timeunit.seconds,
            new synchronousqueue<>(),  // 无界队列，快速响应
            new namedthreadfactory("io-pool")
        );
    }
    // 定时/超时任务：使用scheduledexecutor
    @bean("timeoutpool")
    public scheduledexecutorservice timeoutpool() {
        return executors.newscheduledthreadpool(5, 
            new namedthreadfactory("timeout-pool"));
    }
}