Java 8 CompletableFuture实现多线程批量处理与超时控制

随着业务系统复杂度不断提升，单线程串行执行任务已经难以满足高并发场景下的性能需求。例如批量查询用户信息、同时调用多个远程接口、批量处理订单数据等操作，如果采用传统同步方式处理，会导致大量时间消耗在等待过程中。

Java 8引入的CompletableFuture为异步编程提供了更加灵活的解决方案，它不仅能够实现多线程任务并行执行，还支持任务组合、异常处理以及超时控制。合理使用CompletableFuture，可以有效提升系统吞吐量和响应速度。

CompletableFuture简介

CompletableFuture位于java.util.concurrent包中，是Java 8提供的异步任务处理工具类。它实现了Future接口，同时扩展了Future的能力，使开发人员能够更加方便地编排多个异步任务。

传统Future存在一些不足：

获取结果时必须主动调用get()方法阻塞等待；
多个异步任务之间难以组合；
缺少完善的异常处理机制；
无法方便实现任务完成后的回调操作。

CompletableFuture通过链式API解决了这些问题，可以将多个异步任务按照业务流程进行组合。

使用CompletableFuture实现多线程批量处理

假设系统需要批量查询多个商品信息，如果逐个查询，每次请求都需要等待返回结果，总耗时会随着数据量增加而增长。

传统方式：

for (Long id : productIds) {
    Product product = queryProduct(id);
    result.add(product);
}

这种方式执行效率较低，因为每一次查询都必须等待上一次完成。

使用CompletableFuture后，可以让多个任务同时执行。

创建异步任务

CompletableFuture提供了supplyAsync()方法执行带返回值的异步任务。

示例：

public CompletableFuture queryAsync(Long id) {
    return CompletableFuture.supplyAsync(() -> {
        return queryProduct(id);
    });
}

调用supplyAsync后，任务会提交到默认线程池ForkJoinPool中执行。

批量提交任务

批量处理时，可以将多个CompletableFuture放入集合中统一管理。

List productIds = Arrays.asList(1L, 2L, 3L, 4L);

List> futures = productIds.stream()
        .map(this::queryAsync)
        .collect(Collectors.toList());

此时多个商品查询任务已经并行执行。

等待所有任务完成

CompletableFuture提供allOf()方法，可以等待多个异步任务全部完成。

CompletableFuture allFuture =
        CompletableFuture.allOf(
            futures.toArray(new CompletableFuture[0])
        );

allFuture.join();

任务完成后，可以获取每个异步结果：

List products = futures.stream()
        .map(CompletableFuture::join)
        .collect(Collectors.toList());

相比传统循环查询，多任务并行可以明显降低整体响应时间。

自定义线程池提升任务控制能力

虽然CompletableFuture默认使用ForkJoinPool，但在生产环境中通常不建议直接使用默认线程池。

原因包括：

默认线程数量不可控；
可能与系统其他异步任务产生资源竞争；
难以根据业务调整并发能力。

推荐自定义线程池：

ExecutorService executor =
        Executors.newFixedThreadPool(10);

CompletableFuture future =
        CompletableFuture.supplyAsync(() -> {
            return queryProduct(1001L);
        }, executor);

通过自定义线程池，可以控制：

最大并发数量；
任务队列长度；
线程生命周期；
系统资源占用。

在高并发业务中，线程池参数需要结合CPU核心数、任务类型以及接口响应时间进行调整。

CompletableFuture实现任务超时控制

在实际应用中，异步任务可能由于网络异常、数据库响应慢等原因长时间无法完成。如果没有超时机制，会导致线程长期占用。

Java 8中的CompletableFuture本身没有直接提供completeOnTimeout()方法，这些方法是在Java 9之后新增的。因此Java 8项目通常需要通过其他方式实现超时控制。

使用ScheduledExecutorService实现超时

一种常见方式是结合定时线程池。

示例：

public static  CompletableFuture timeoutFuture(
        CompletableFuture future,
        long timeout,
        TimeUnit unit) {

    ScheduledExecutorService scheduler =
            Executors.newScheduledThreadPool(1);

    scheduler.schedule(() -> {
        future.completeExceptionally(
            new TimeoutException("任务执行超时")
        );
    }, timeout, unit);

    return future;
}

使用：

CompletableFuture future =
        CompletableFuture.supplyAsync(() -> {
            return queryProduct(1001L);
        });

timeoutFuture(future, 3, TimeUnit.SECONDS)
        .exceptionally(e -> {
            System.out.println(e.getMessage());
            return null;
        });

当任务超过指定时间仍未完成时，会触发异常处理逻辑。

CompletableFuture异常处理

批量任务执行过程中，某一个任务失败是非常常见的情况。如果没有正确处理异常，可能导致整个批处理流程失败。

exceptionally处理异常

CompletableFuture future =
        CompletableFuture.supplyAsync(() -> {
            throw new RuntimeException("查询失败");
        })
        .exceptionally(e -> {
            return "默认结果";
        });

exceptionally可以在发生异常时返回备用结果。

handle统一处理结果和异常

future.handle((result, exception) -> {

    if (exception != null) {
        return "失败";
    }

    return result;

});

handle比exceptionally更加灵活，因为它可以同时处理成功和失败情况。

使用thenCombine组合多个任务

很多业务场景需要等待多个异步任务完成后进行下一步处理。

例如：

查询用户基础信息；
查询用户订单数据；
合并生成用户详情。

示例：

CompletableFuture userFuture =
        CompletableFuture.supplyAsync(() -> queryUser());

CompletableFuture> orderFuture =
        CompletableFuture.supplyAsync(() -> queryOrders());


CompletableFuture detailFuture =
        userFuture.thenCombine(orderFuture,
        (user, orders) -> {
            return new UserDetail(user, orders);
        });

thenCombine可以将两个独立任务的结果组合起来。