当前位置：首页 > news >正文

CompletableFuture 使用和源码解读

news 来源：原创 2025/5/20 23:04:49

引言

CompletableFuture 是 Java 8 引入的一个强大的异步编程工具，用于处理异步操作和处理结果。它实现了 Future 和 CompletionStage 接口，提供了丰富的方法来处理异步任务的完成、组合和异常处理。

CompletableFuture本质是对异步线程的返回值的处理，所以要有线程池和对异步结果的处理

方法使用

使用 runAsync 执行无返回值的异步任务

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CompletableFutureExample {
    public static void main(String[] args) {
        // 创建一个线程池
        ExecutorService executor = Executors.newSingleThreadExecutor();

        // 创建一个无返回值的异步任务
        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
            try {
                Thread.sleep(1000);
                System.out.println("异步任务执行完成");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, executor);

        // 主线程继续执行其他任务
        System.out.println("主线程继续执行");

        // 等待异步任务完成
        future.join();

        // 关闭线程池
        executor.shutdown();
    }
}

使用 supplyAsync 执行有返回值的异步任务

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CompletableFutureSupplyExample {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newSingleThreadExecutor();

        // 创建一个有返回值的异步任务
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1000);
                return "异步任务的结果";
            } catch (InterruptedException e) {
                e.printStackTrace();
                return null;
            }
        }, executor);

        System.out.println("主线程继续执行");

        // 获取异步任务的结果
        String result = future.join();
        System.out.println("异步任务的结果是: " + result);

        executor.shutdown();
    }
}

和runAsync区别是join方法有没有返回结果

使用 thenApply 对结果进行转换

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ThenApplyExample {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newSingleThreadExecutor();

        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1000);
                return 10;
            } catch (InterruptedException e) {
                e.printStackTrace();
                return 0;
            }
        }, executor).thenApply(result -> result * 2);

        Integer finalResult = future.join();
        System.out.println("最终结果是: " + finalResult);

        executor.shutdown();
    }
}

使用 thenAccept 消费结果

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ThenAcceptExample {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newSingleThreadExecutor();

        CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1000);
                return 20;
            } catch (InterruptedException e) {
                e.printStackTrace();
                return 0;
            }
        }, executor).thenAccept(result -> System.out.println("消费结果: " + result));

        future.join();
        executor.shutdown();
    }
}

源码解读

CompletableFuture 内部有一个 volatile 类型的 state 变量来表示任务的状态，常见的状态有：

NEW：初始状态，表示任务还未开始执行。
COMPLETING：正在完成状态，表示任务正在执行完成操作。
NORMAL：正常完成状态，表示任务正常执行并返回结果。
EXCEPTIONAL：异常完成状态，表示任务执行过程中抛出了异常。

runAsync

public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor) {
    return asyncRunStage(screenExecutor(executor), runnable);
}

private static CompletableFuture<Void> asyncRunStage(Executor e, Runnable f) {
    if (f == null) throw new NullPointerException();
    CompletableFuture<Void> d = new CompletableFuture<Void>();
    e.execute(new AsyncRun(d, f));
    return d;
}

runAsync 方法接收一个 Runnable 任务和一个 Executor 线程池，它会创建一个新的 CompletableFuture 对象，并将任务封装成 AsyncRun 对象提交给线程池执行。

supplyAsync 方法源码

public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor) {
    return asyncSupplyStage(screenExecutor(executor), supplier);
}

private static <U> CompletableFuture<U> asyncSupplyStage(Executor e, Supplier<U> f) {
    if (f == null) throw new NullPointerException();
    CompletableFuture<U> d = new CompletableFuture<U>();
    e.execute(new AsyncSupply<U>(d, f));
    return d;
}

supplyAsync 方法与 runAsync 类似，只不过它接收的是一个 Supplier 任务，该任务有返回值。它会创建一个新的 CompletableFuture 对象，并将任务封装成 AsyncSupply 对象提交给线程池执行。

thenApply 方法源码

public <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn) {
    return uniApplyStage(null, fn);
}

private <V> CompletableFuture<V> uniApplyStage(Executor e, Function<? super T,? extends V> f) {
    if (f == null) throw new NullPointerException();
    CompletableFuture<V> d = new CompletableFuture<V>();
    if (e != null || !d.uniApply(this, f, null)) {
        UniApply<T,V> c = new UniApply<T,V>(e, d, this, f);
        push(c);
        c.tryFire(SYNC);
    }
    return d;
}

thenApply 方法源码thenApply 方法接收一个 Function 函数，用于对 CompletableFuture 的结果进行转换。它会创建一个新的 CompletableFuture 对象，并将转换操作封装成 UniApply 对象。如果当前任务已经完成，则直接执行转换操作；否则，将 UniApply 对象添加到等待队列中，等待当前任务完成后再执行转换操作。