java - Patterns.ask 超时后消息将转到哪里？

Question

最近在 Akka actor 中创建的等待处理的 scala.concurrent.Future 对象遇到了一些超时，我想知道如何处理这些超时事件。他们真的迷路了吗？它们是否会重试并保存在内存中，或者它是如何工作的？

为了提供一些上下文，代码如下。

List<Future<MyMessage>> futureMessageList = plainMessages.stream()
    .map(this::toFuture)
    .collect(Collectors.toList());    

Futures.sequence(futureMessageList, ExecutionContexts.global())
                .onComplete(new OnComplete<Iterable<MyMessage>>() {
                    @Override
                    public void onComplete(Throwable throwable, Iterable<MyMessage> messages) {
                        ... // iterate futureMessageList list

在onComplete内futureMessageList 的迭代发生，基本上由 Future 组成封装 MyMessage 的对象.

但是，函数 toFuture做Patterns.ask()对于给定的调度程序，这似乎花费的时间超过了我发送的超时时间(60 秒)。请考虑到响应时间取决于底层系统，该系统可能处于高负载或没有最快的网络，具体取决于其运行的环境。

Future<MyMessage> message = Patterns.ask(actorSystem.getSampleDispatcher(), msg, TIMEOUT_60_SECS)

所以我的问题是，在onComplete之后由于 Future 没有及时处理，抛出以下异常...

java.lang.NullPointerException
    at my.package.Clazz.onComplete(Clazz.java:4)
    at my.package.Clazz$1.onComplete(Clazz.java:5)
    at akka.dispatch.OnComplete.internal(Future.scala:258)
    at akka.dispatch.OnComplete.internal(Future.scala:256)
    at akka.dispatch.japi$CallbackBridge.apply(Future.scala:186)
    at akka.dispatch.japi$CallbackBridge.apply(Future.scala:183)
    at scala.concurrent.impl.CallbackRunnable.run$$$capture(Promise.scala:32)
    at scala.concurrent.impl.CallbackRunnable.run(Promise.scala)
    at scala.concurrent.impl.ExecutionContextImpl$AdaptedForkJoinTask.exec(ExecutionContextImpl.scala:121)

是那些MyMessage objs 保存在内存中并随后重试？我应该以某种方式处理异常并使用内存列表处理这些超时消息，或者我应该如何解决这个问题？

Answer 1

何时 ask由于未收到回复而超时，它完成了 Future (或 CompletionStage )失败。该消息可能仍在某个地方正在处理，如果有响应，它将以死信结束( https://doc.akka.io/docs/akka/current/general/message-delivery-reliability.html#dead-letters )。其他可能发生超时的情况是，如果执行者已停止或崩溃处理消息、请求或响应丢失(除非响应执行者是远程的，否则不太可能)。

Future.sequence当传递给它的所有 future 都成功完成时，它将成功完成；如果其中任何一个失败，则失败。

这意味着如果任何请求超时，您将得到 null作为 messages 参数和来自第一个失败的 future 的异常作为 throwable onComplete 中的参数回调。

如果您想获取部分结果列表，每个结果要么是成功值，要么是异常。您可以在将每个 future 传递给 Future.sequence 之前借助恢复来做到这一点。 .