Java Actor模型下如何用CompletionStage实现非阻塞调用替代get()阻塞逻辑

核心前提 #

Actor模型禁止阻塞调度线程，异步结果无法同步返回，原有foo()方法的Map<Integer, Integer>返回类型必须调整为CompletionStage<Map<Integer, Integer>>，否则必然需要阻塞等待结果，无法实现真正的非阻塞。

代码实现 #

基础改造（通用CompletionStage用法） #

import java.util.Map;
import java.util.List;
import java.util.concurrent.CompletionStage;

// 改造后的foo方法，返回异步包装的结果
CompletionStage<Map<Integer, Integer>> foo() {
    CompletionStage<List<String>> listStage = foo_2();
    // 通过thenApply承接异步返回的List，执行业务转换逻辑，和原阻塞逻辑的处理完全一致
    return listStage.thenApply(finalList -> {
        // 此处直接复用你原有基于finalList生成Map的业务代码，不需要修改
        Map<Integer, Integer> finalMap = /* 原有生成Map的逻辑 */;
        return finalMap;
    });
}

CompletionStage<List<String>> foo_2() {
    // 直接返回ask得到的CompletionStage，不需要调用get()阻塞
    return ask(Actor2);
}

Actor场景适配（解决线程安全问题） #

由于Actor的内部状态只能由自身调度线程修改，需要保证回调逻辑在Actor的调度线程中执行，避免并发问题。如果你使用的是Akka等Actor框架，可参考如下用法：

// Actor消息处理逻辑示例
@Override
public Receive createReceive() {
    return receiveBuilder()
        .match(GetMapRequest.class, req -> {
            ActorRef requester = getSender();
            // 触发异步处理流程
            CompletionStage<Map<Integer, Integer>> resultStage = foo();
            // 指定回调使用Actor的调度线程执行
            resultStage.thenAcceptAsync(finalMap -> {
                // 回复请求方结果，逻辑和原阻塞实现完全一致
                requester.tell(new GetMapResponse(finalMap), getSelf());
            }, getContext().getDispatcher())
            // 异常处理，和原get()抛出异常的处理逻辑对齐
            .exceptionally(ex -> {
                requester.tell(new ProcessFailedException(ex), getSelf());
                return null;
            });
            // 此处直接返回，不需要等待结果，完全不阻塞Actor线程
        })
        .build();
}

异常对齐方案 #

如果需要和原有get()抛出异常的逻辑完全一致，可通过exceptionally或者whenComplete捕获处理异常：

CompletionStage<Map<Integer, Integer>> foo() {
    return foo_2()
        .thenApply(finalList -> {
            // 业务转换逻辑
            Map<Integer, Integer> finalMap = /* 原有逻辑 */;
            return finalMap;
        })
        .exceptionally(ex -> {
            // 此处可以按原有逻辑处理异常，比如封装自定义异常抛出，或者返回默认值
            throw new BizException("查询Actor2失败", ex);
        });
}

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