Akka.NET 有效地查询参与者

Question

我正在使用 Akka.NET 为生产项目创建概念证明，但我遇到了查询概念理解问题。

情况如下： CoordinatorActor 有数千个Hotel-Actors 的列表。

我想查询所有Hotel-Actors 在特定日期有空房的所有酒店。

当然，我可以通过它们进行查询并发送.Ask<> 请求以获取特定日期。持有所有任务的参考并做一个Task.WhenAll(requests)。但这感觉有点不自然。

或者，我可以一次向所有酒店（ActorSelection 或路由器）发送一条包含特定日期请求的广播消息，但我不知道他们何时所有回复了 @ 987654325@留言。

有没有人建议如何解决这个问题？

Answer 1

是的，你的感受就在这里。在参与者之间使用 Ask 进行通信被认为是非常低效的 - 主要是因为每个 ask 需要分配单独的消息侦听器。

第一个好问题是：您需要等待他们都回复吗？也许响应可以在它们到来时发出。

如果您需要在回复之前收集所有数据，您需要有一些方法来计算所有消息，以保证其中一些消息仍然处于待处理状态 - 在这种情况下使用 ActorSelection 是不可行的.您需要与每条消息相关联的计数器或标识符列表 - 虽然它们甚至可以是普通数字，但通常IActorRefs 更易于使用。

您可以在下面看到可以为此特定任务创建的 Aggregator 演员的简化示例 - 一旦没有更多消息等待或超时，它将自动返回所有回复，它收到，并自行停止发生了。

class Aggregator<T> : ReceiveActor
{
    private IActorRef originalSender;
    private ISet<IActorRef> refs;

    public Aggregator(ISet<IActorRef> refs)
    {
        this.refs = refs;
        // this operation will finish after 30 sec of inactivity
        // (when no new message arrived)
        Context.SetReceiveTimeout(TimeSpan.FromSeconds(30));
        ReceiveAny(x =>
        {
            originalSender = Sender;
            foreach (var aref in refs) aref.Tell(x);
            Become(Aggregating);
        });
    }

    private void Aggregating()
    {
        var replies = new List<T>();
        // when timeout occurred, we reply with what we've got so far
        Receive<ReceiveTimeout>(_ => ReplyAndStop(replies));
        Receive<T>(x =>
        {
            if (refs.Remove(Sender)) replies.Add(x);
            if (refs.Count == 0) ReplyAndStop(replies);
        });
    }

    private void ReplyAndStop(List<T> replies)
    {
        originalSender.Tell(new AggregatedReply<T>(replies));
        Context.Stop(Self);
    }
}