Akka 的 Java 聚合器

Question

我正在尝试实现 Java Aggregator对于 Akka ，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 提供了一个功能，当特定参与者在预定义的时间内没有收到任何内容时，就会超时。

在 Java 中，你可以在 actor 中执行类似的操作:

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

当此触发器时，它会向参与者发送一条 ReceiveTimeout 类型的消息，然后您可以决定要执行的操作(异常、日志记录、重置...)。

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

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