Akka 的 Java 聚合器

Question

我正在尝试为 Akka 实现 Java Aggregator，since it doesn't look like Java API 支持它们（为什么不！？）

这是我迄今为止最好的尝试：

// Groovy pseudo-code
abstract class Aggregator<T> extends UntypedActor {
    ActorRef recipient
    Set<T> aggregation
    // TODO: Timer timer (?)

    abstract boolean isAggregated()

    @Override
    void onReceive(Object message) {
        aggregation << message as T

        if(isAggregated()) {
            recipient.tell(new Aggregation(aggregation))    // again, pseudo-code
            aggregation.clear()
            // TODO: timer.reset()
        }
    }
}

缺少的是某种Timer 构造，它将在Aggregator 尚未聚合的情况下在60 秒后超时。超时时，它应该抛出某种异常。在聚合时，应重置计时器。任何想法如何做到这一点？

Answer 1

您正在寻找的是ReceiveTimeout。 Akka 提供了一个功能，当特定的 Actor 在预定义的时间内没有收到任何东西时，它会超时。

在 Java 中，你会在你的 actor 内部做这样的事情：

getContext().setReceiveTimeout(Duration.create("1 second"));

当这个触发器发送一个ReceiveTimeout类型的消息给actor，然后你可以决定你想要做什么（异常、日志记录、重置......）。

您可以在“接收超时”部分下找到更多信息：http://doc.akka.io/docs/akka/snapshot/java/untyped-actors.html

另一方面，在 github 中有一些开源库可以做这些事情。更多示例请查看https://github.com/sksamuel/akka-patterns。