如何使用5个CompletableFuture异步执行20个Runnable任务？

嘿，这个问题其实很好解决——核心思路就是把20个任务分摊到5个CompletableFuture里，让它们异步处理所有任务。下面给你两种实用的实现方式，你可以根据自己的需求选择：

方式一：固定分组串行执行 #

如果任务之间不需要动态调度，只想简单把20个任务平均分给5个CompletableFuture（每个Future负责4个任务），可以用这种分组串行的方式：

import java.util.Random;
import java.util.stream.DoubleStream;
import java.util.concurrent.CompletableFuture;

public class CompletableFutureBatchDemo {
    public static void main(String[] args) {
        Runnable task = () -> {
            long startTime = System.currentTimeMillis();
            Random random = new Random();
            while (System.currentTimeMillis() - startTime < 3000) {
                DoubleStream.generate(() -> random.nextDouble())
                        .limit(random.nextInt(100))
                        .map(Math::cos)
                        .sum();
            }
            System.out.println("Task finished by thread: " + Thread.currentThread().getName());
        };

        final int futureCount = 5;
        final int totalTasks = 20;
        int tasksPerFuture = totalTasks / futureCount;

        // 创建5个CompletableFuture，每个执行4个任务
        CompletableFuture<Void>[] futures = new CompletableFuture[futureCount];
        for (int i = 0; i < futureCount; i++) {
            futures[i] = CompletableFuture.runAsync(() -> {
                for (int j = 0; j < tasksPerFuture; j++) {
                    task.run();
                }
            });
        }

        // 等待所有异步任务完成
        CompletableFuture.allOf(futures).join();
        System.out.println("All 20 tasks are done!");
    }
}

关键说明： #

• 我们先计算每个Future要处理的任务数：20 ÷ 5 = 4，保证任务平均分配。
• 用CompletableFuture.runAsync()创建异步任务，每个Future内部会串行执行自己的4个任务。
• 最后调用allOf().join()确保所有异步任务都执行完毕后，再继续后续逻辑。

方式二：动态任务调度（更高效） #

如果你的任务执行时间不稳定（有的快有的慢），想让空闲的线程自动接手新任务，避免资源浪费，可以结合阻塞队列来实现动态分配：

import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.stream.DoubleStream;
import java.util.concurrent.CompletableFuture;

public class CompletableFutureQueueDemo {
    public static void main(String[] args) {
        Runnable task = () -> {
            long startTime = System.currentTimeMillis();
            Random random = new Random();
            while (System.currentTimeMillis() - startTime < 3000) {
                DoubleStream.generate(() -> random.nextDouble())
                        .limit(random.nextInt(100))
                        .map(Math::cos)
                        .sum();
            }
            System.out.println("Task finished by thread: " + Thread.currentThread().getName());
        };

        final int futureCount = 5;
        final int totalTasks = 20;

        // 用阻塞队列存放所有任务
        BlockingQueue<Runnable> taskQueue = new ArrayBlockingQueue<>(totalTasks);
        for (int i = 0; i < totalTasks; i++) {
            taskQueue.add(task);
        }

        // 创建5个CompletableFuture，不断从队列取任务执行
        CompletableFuture<Void>[] futures = new CompletableFuture[futureCount];
        for (int i = 0; i < futureCount; i++) {
            futures[i] = CompletableFuture.runAsync(() -> {
                Runnable currentTask;
                // 循环取任务，直到队列空了为止
                while ((currentTask = taskQueue.poll()) != null) {
                    currentTask.run();
                }
            });
        }

        CompletableFuture.allOf(futures).join();
        System.out.println("All 20 tasks are done!");
    }
}

关键说明： #

• 我们把20个任务都放进阻塞队列，每个CompletableFuture会不断从队列里取任务执行。
• 这种方式下，执行完任务的线程会立刻去取下一个任务，任务分配更均衡，不会出现有的线程闲等、有的线程忙死的情况。
• 如果需要自定义线程池，你可以给runAsync()传入自己的Executor，比如：CompletableFuture.runAsync(() -> {...}, new ThreadPoolExecutor(...))。

内容的提问来源于stack exchange，提问作者Nick Kulese