Akka.NET 高效查询参与者 [英] Akka.NET query actors efficiently

问题描述

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

I'm creating a proof of concept with Akka.NET for a production project but I'm facing a query concept understanding problem.

情况如下:CoordinatorActor 有一个包含数千个 Hotel-Actors 的列表.

Situation is a follows: CoordinatorActor has a list of thousands of Hotel-Actors.

我想查询所有 Hotel-Actors 以获取在特定日期有空房的所有酒店.

I would like to query all Hotel-Actors for all hotels with a room available on a specific date.

当然，我可以通过它们进行 foreach 并发送 .Ask<> 请求以获取特定日期.持有所有任务的引用并执行Task.WhenAll(requests).但这感觉有点不自然.

Of course I could foreach through them and sent a .Ask<> request for the specific date. Holding a reference of all the tasks and do a Task.WhenAll(requests). But that feels a little unnatural.

或者，我可以一次向所有酒店(ActorSelection 或路由器)发送请求特定日期的广播消息，但我不知道他们何时全部回复<代码>告诉消息.

Alternatively I could send a broadcast message with a request for the specific date to all hotels at once (ActorSelection or router), but then I don't know when they all responded back with a Tell message.

有人建议如何解决这个问题吗?

Does anyone has a suggestion how to solve this?

推荐答案

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

Yes, your feelings are right here. Using Ask for communication between actors is considered highly inefficient - mostly because of each ask needs to allocate separate message listener.

第一个好问题是:您是否需要等待他们全部回复?也许可以在响应来时发出响应.

First good question would be: do you need to wait for them all to respond back? Maybe responses can be emitted as they come.

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

In case, when you need to collect all of the data before replying, you need to have some way of counting all messages in order to guarantee if some of them are still pending - in that case using ActorSelection is not feasible. You'll need counter or list of identifiers that could be associated with each message - while they can be even ordinary numbers, usually IActorRefs are easier to work with.

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

Below you can see simplified example of Aggregator actor that can be created for this particular task - it will automatically return all replies, it received, and stop itself, once there are no more messages to wait for or a timeout has occurred.

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);
    }
}

这篇关于Akka.NET 高效查询参与者的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！