在当前参与者进程中使用参与者的响应

Question

我对如何以非阻塞方式解决这种情况感到困惑。

考虑两个演员 Actor1 和 Actor2

Actor1内

Map<Int, Int> foo() {
     List<String> finalList = foo_2();
     Map<Int, Int> finalMap = // do stuff with finalList to get Map<Int, Int>;

     return finalMap;
}

List<String> foo_2() {
    
     CompletableFuture<List<String>> Querylist = ask(Actor2)
     Querylist.get();
     
     return QueryList;
}

当前在 foo_2 中，Querylist.get() 是一个阻塞调用。我想以某种非阻塞方式解决这个问题。我在Actor1 内部为Actor2 创建了一个消息适配器，因此Actor2 发送的任何消息都将由Actor1 处理。

我使用下面的方法来修改阻塞调用

Map<Int, Int> foo() {
     CompletionStage<List<String>> finalList = foo_2();
     finalList.whenComplete(
        // what to do here? 
     )
     // Map<Int, Int> finalMap = // do stuff with finalList to get Map<Int, Int>;

     return finalMap;
}

CompletionStage<List<String>> foo_2() {
    
     CompletionStage<List<String>> Querylist = ask(Actor2)
     
     
     return QueryList;
}

我不确定如何正确使用 CompletionStage 构造来获得与阻塞 futures.get() 调用相同的结果。

Answer 1

如果您使用的是 Akka Typed（从标签中暗示），则根本不需要 future 或消息适配器。只需使用ActorContext.ask。

请参阅documentation for request-response with ask between two actors。

一般来说，您将摆脱foo 和foo_2 方法，将您设置的消息适配器移动到ActorContext.ask 调用中，并替换您之前调用foo 的实例致电ActorContext.ask。如果您的参与者对导致请求的消息的回复取决于对请求的响应，那么一个好的做法是将所需的状态部分嵌入到适配器生成的消息中。

Answer 2

您可以使用pipeToSelf（参见https://doc.akka.io/docs/akka/current/typed/interaction-patterns.html#send-future-result-to-self）将请求的结果发送给参与者本身。与其尝试直接在 foo() 中获取 finalList 的值（这只能通过阻塞 get 实现），不如将 foo() 的结果发送给 actor 本身，在这种情况下，您可以像处理任何其他消息。最好为此创建特定的消息类型。

您还应该看看CompletionStage 方法，最重要的是thenApply (https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletionStage.html#thenApply-java.util.function.Function-) 可以转换结果，例如从finalList 创建一个Map，例如来自Map 的MapMessage。然后，您将像处理 actor 中的任何其他消息一样处理 MapMessage。