文章/答案/技术大牛

发布

社区首页 >专栏 >并发设计模式实战系列(15)：Future/Promise

并发设计模式实战系列(15)：Future/Promise

摘星.

发布于 2025-05-20 07:06:14

11100

代码可运行

文章被收录于专栏：博客专享博客专享

运行总次数：0

代码可运行

🌟 大家好，我是摘星！ 🌟

今天为大家带来的是并发设计模式实战系列，第十五章Future/Promise，废话不多说直接开始~

一、核心原理深度拆解

1. 异步计算双阶段模型

┌─────────────┐    ┌─────────────┐    ┌─────────────┐
│   Task      │───>│   Future    │───>│   Callback  │
│ Submission  │<───│  (Promise)  │<───│  Execution  │
└─────────────┘    └─────────────┘    └─────────────┘

提交阶段：主线程提交任务后立即返回Future占位符
计算阶段：工作线程异步执行计算，通过Promise设置结果
回调阶段：结果就绪后触发回调（观察者模式）

2. 状态机流转

public interface Future<V> {
    boolean isDone();      // 完成状态（成功/失败/取消）
    V get() throws...;     // 阻塞获取结果
    void addCallback(...); // 回调注册
}

二、生活化类比：快递柜取件

系统组件	现实类比	核心行为
Future	快递柜取件码	凭码查询包裹是否到达
Promise	快递员存件操作	实际将包裹放入柜中并更新状态
Callback	短信通知服务	包裹入柜后自动发送取件提醒

异步流程：下单→获得取件码（Future）→快递员送货（异步计算）→短信通知（Callback）

三、Java代码实现（生产级Demo）

1. 完整可运行代码

import java.util.concurrent.*;
import java.util.function.Consumer;

public class FuturePromiseDemo {

    // 1. 自定义Promise实现
    static class MyPromise<V> implements Future<V>, Runnable {
        private volatile V result;
        private volatile Throwable error;
        private volatile boolean isDone;
        private final CountDownLatch latch = new CountDownLatch(1);
        private final List<Consumer<V>> callbacks = new CopyOnWriteArrayList<>();

        // 提交任务时执行的方法
        @Override
        public void run() {
            try {
                // 模拟耗时计算
                Thread.sleep(1000);
                setResult((V) "计算结果"); // 实际业务逻辑替换此处
            } catch (Exception e) {
                setError(e);
            }
        }

        // Promise核心方法：设置结果
        public void setResult(V result) {
            this.result = result;
            this.isDone = true;
            latch.countDown();
            notifyCallbacks();
        }

        // Promise核心方法：设置异常
        public void setError(Throwable error) {
            this.error = error;
            this.isDone = true;
            latch.countDown();
        }

        private void notifyCallbacks() {
            callbacks.forEach(cb -> cb.accept(result));
        }

        // Future实现方法
        @Override
        public V get() throws InterruptedException, ExecutionException {
            latch.await();
            if (error != null) throw new ExecutionException(error);
            return result;
        }

        @Override
        public boolean isDone() {
            return isDone;
        }

        // 注册回调（非JUC标准方法）
        public void addCallback(Consumer<V> callback) {
            if (isDone) {
                callback.accept(result);
            } else {
                callbacks.add(callback);
            }
        }
    }

    // 2. 使用示例
    public static void main(String[] args) throws Exception {
        ExecutorService executor = Executors.newCachedThreadPool();
        
        // 创建Promise并提交任务
        MyPromise<String> promise = new MyPromise<>();
        executor.submit(promise);

        // 注册回调
        promise.addCallback(result -> 
            System.out.println("[回调] 异步结果: " + result));

        // 同步阻塞获取
        System.out.println("[主线程] 立即返回，继续其他工作...");
        System.out.println("最终结果: " + promise.get());

        executor.shutdown();
    }
}

2. 关键机制说明

// 1. 状态同步控制
private volatile boolean isDone;  // 保证可见性
private final CountDownLatch latch; // 实现阻塞等待

// 2. 线程安全回调列表
private final List<Consumer<V>> callbacks = new CopyOnWriteArrayList<>();

// 3. 异常处理流程
public void setError(Throwable error) {
    this.error = error;
    this.isDone = true;
    latch.countDown(); // 唤醒所有等待线程
}

四、横向对比表格

1. 异步模式对比

模式	核心特点	适用场景	Java实现类
Future	阻塞式获取结果	简单异步任务	FutureTask
CompletableFuture	链式调用+组合操作	复杂异步流水线	CompletableFuture
Promise	可写的结果容器	跨线程结果传递	需自行实现
Callback	事件驱动无阻塞	高并发IO	Netty的ChannelFuture

2. 回调注册方式对比

方法	触发时机	线程安全性	链式支持
addCallback	结果就绪后立即执行	需自行保证	不支持
thenApply	前序阶段完成后触发	内置线程池控制	支持
whenComplete	无论成功失败都执行	可能在不同线程执行	支持

五、高级应用技巧

1. 组合多个异步任务

CompletableFuture<String> query1 = queryDatabase("sql1");
CompletableFuture<String> query2 = queryDatabase("sql2");

// 并行执行后合并结果
CompletableFuture<String> merged = query1.thenCombineAsync(query2, 
    (r1, r2) -> r1 + "|" + r2,
    ForkJoinPool.commonPool());

2. 超时控制

Future<String> future = executor.submit(task);
try {
    String result = future.get(2, TimeUnit.SECONDS);
} catch (TimeoutException e) {
    future.cancel(true); // 中断任务执行
}

3. 回调线程控制

promise.addCallback(result -> {
    // 指定回调执行线程池
    ForkJoinPool.commonPool().execute(() -> processResult(result));
});

通过这种 原理+实现+对比 的立体解析，可以掌握：

Future/Promise的双阶段异步本质
如何实现生产级的Promise容器
不同异步模式的适用场景选择
复杂场景下的组合使用技巧

六、源码级实现剖析（接五）

1. JDK FutureTask 核心逻辑

// 状态机定义（OpenJDK 17）
private volatile int state;
static final int NEW          = 0; // 初始化状态
static final int COMPLETING   = 1; // 临时状态
static final int NORMAL       = 2; // 正常完成
static final int EXCEPTIONAL  = 3; // 异常完成
static final int CANCELLED    = 4; // 已取消
static final int INTERRUPTING = 5; // 中断中
static final int INTERRUPTED  = 6; // 已中断

// 结果存储设计
private Object outcome; // 非volatile，依赖状态可见性保证

2. CompletableFuture 回调链实现

// 回调节点结构（简化版）
static final class UniCompletion<T,V> extends Completion {
    Executor executor;         // 执行线程池
    CompletableFuture<V> dep;   // 依赖的前序Future
    BiFunction<? super T,? super Throwable,? extends V> fn; // 回调函数

    void tryFire(int mode) {    // 触发回调执行
        if (dep != null && 
            compareAndSetState(0, 1)) { // CAS保证线程安全
            fn.apply(src, ex);  // 实际执行用户回调
        }
    }
}

七、生产环境最佳实践

1. 异常处理模板

CompletableFuture.supplyAsync(() -> {
        // 业务逻辑
        return doSomething();
    })
    .exceptionally(ex -> {      // 捕获所有异常
        log.error("任务失败", ex);
        return defaultValue;    // 提供降级值
    })
    .thenAccept(result -> {     // 只处理成功情况
        updateUI(result); 
    });

2. 资源清理策略

ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
try {
    CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
        // 使用try-with-resources确保资源释放
        try (Connection conn = getConnection()) {
            process(conn);
        }
    }, executor);
    
    future.whenComplete((r, ex) -> {
        if (ex != null) {
            cleanupTempFiles(); // 失败时清理临时文件
        }
    });
} finally {
    executor.shutdown(); // 确保线程池关闭
}

3. 性能监控指标

// 监控Future完成时长
Timer.Sample sample = Timer.start();
future.whenComplete((r, ex) -> {
    sample.stop(registry.timer("async.task.time"));
});

// 监控队列积压
ThreadPoolExecutor pool = (ThreadPoolExecutor) executor;
metrics.gauge("task.queue.size", pool.getQueue()::size);

八、与其他模式的协作

1. 结合发布-订阅模式

EventBus bus = new EventBus();
CompletableFuture.supplyAsync(() -> fetchData())
    .thenAccept(data -> {
        bus.post(new DataReadyEvent(data)); // 异步事件通知
    });

// 订阅方处理
@Subscribe
void handleDataReady(DataReadyEvent event) {
    // 处理已完成的数据
}

2. 与反应式编程整合

// CompletableFuture -> Mono
Mono.fromFuture(() -> {
    return CompletableFuture.supplyAsync(() -> {
        return reactiveDao.query();
    });
}).subscribeOn(Schedulers.boundedElastic())
  .subscribe(System.out::println);

// Mono -> CompletableFuture
reactorMono.toFuture().thenApply(...);

九、各语言实现对比

语言	核心实现类	特色功能	典型使用场景
Java	CompletableFuture	链式组合、CompletionStage	服务端异步编排
C#	Task	async/await语法糖	UI线程非阻塞调用
JavaScript	Promise	then/catch链式调用	前端API请求
Python	asyncio.Future	协程集成	爬虫/高并发IO
Go	chan	通道原生支持	高并发微服务

十、常见陷阱与解决方案

1. 回调地狱问题

反模式：

future.thenApply(r1 -> {
    future2.thenApply(r2 -> {
        future3.thenApply(r3 -> {  // 嵌套层次过深
            return r1 + r2 + r3;
        });
    });
});

解决方案：

// 使用组合式编程
CompletableFuture.allOf(future1, future2, future3)
    .thenApply(v -> {
        return future1.join() + 
               future2.join() + 
               future3.join();
    });

2. 线程泄漏场景

问题代码：

ExecutorService executor = Executors.newFixedThreadPool(5);
CompletableFuture.runAsync(() -> {
    while (true) {  // 无限循环任务
        process();
    }
}, executor);  // 线程永远无法回收

正确做法：

// 使用守护线程或超时控制
ExecutorService executor = Executors.newFixedThreadPool(5, r -> {
    Thread t = new Thread(r);
    t.setDaemon(true);  // 设置为守护线程
    return t;
});

3. 上下文丢失问题

问题现象：

SecurityContext ctx = getContext();
CompletableFuture.runAsync(() -> {
    // 此处无法获取原始上下文
    doPrivilegedAction(); 
}, executor);

解决方案：

// 使用ContextPropagator
ExecutorService wrappedExecutor = ContextPropagator.wrap(executor);
CompletableFuture.runAsync(() -> {
    // 可以获取原始上下文
    doPrivilegedAction();
}, wrappedExecutor);