如何用 Akka.Net 做 WaitAll？

Question

我在 Akka.Net 中有一个演员层次结构，我想知道我是否选择了正确的方法来做某事，或者是否有更好/更简单的方法来实现我想要的。

我的具体示例是，我正在构造一个 User 演员以响应用户登录系统，并且在构造该演员时，我需要两条数据来完成演员的建设。

如果这是常规的 .NET 代码，我可能会有类似以下内容...

public Task<User> LoadUserAsync (string username)
{
  IProfileService profileService = ...;
  IMessageService messageService = ...;

  var loadProfileTask = profileService.GetUserProfileAsync(username);
  var loadMessagesTask = messageService.GetMessagesAsync(username);

  Task.WaitAll(loadProfileTask, loadMessagesTask);

  // Now construct the user from the result of both tasks
  var user = new User
  {
    Profile = loadProfileTask.Result,
    Messages = loadMessagesTask.Result
  }

  return Task.FromResult(user);
}

这里我使用WaitAll等待下级任务完成，让它们并发运行。

我的问题是 - 如果我想在 Akka.Net 中做同样的事情，以下是最常见的方法吗？如图所示，我创建了以下...

当我创建我的用户角色时，我会构建一个（临时）用户加载器角色，它的工作是通过调用配置文件角色和消息角色来获取完整的用户详细信息。获取数据的叶子actor如下...

public class UserProfileLoader : ReceiveActor
{
    public UserProfileLoader()
    {
        Receive<LoadUserRequest>(msg =>
        {
            // Load the user profile from somewhere
            var profile = new UserProfile();

            // And respond to the Sender
            Sender.Tell(profile);
            Self.Tell(PoisonPill.Instance);
        });
    }
}

public class UserMessagesLoader : ReceiveActor
{
    public UserMessagesLoader()
    {
        Receive<LoadUserRequest>(msg =>
        {
            // Load the messages from somewhere
            var messages = new List<Message>();

            // And respond to the Sender
            Sender.Tell(messages);
            Self.Tell(PoisonPill.Instance);
        });
    }
}

他们从哪里得到这个讨论的数据并不重要，但两者都只是通过返回一些数据来响应请求。

然后我有协调两个数据收集参与者的参与者......

public class UserLoaderActor : ReceiveActor
{
    public UserLoaderActor()
    {
        Receive<LoadUserRequest>(msg => LoadProfileAndMessages(msg));
        Receive<UserProfile>(msg =>
        {
            _profile = msg;
            FinishIfPossible();
        });

        Receive<List<Message>>(msg =>
        {
            _messages = msg;
            FinishIfPossible();
        });
    }

    private void LoadProfileAndMessages(LoadUserRequest msg)
    {
        _originalSender = Sender;
        Context.ActorOf<UserProfileLoader>().Tell(msg);
        Context.ActorOf<UserMessagesLoader>().Tell(msg);
    }

    private void FinishIfPossible()
    {
        if ((null != _messages) && (null != _profile))
        {
            _originalSender.Tell(new LoadUserResponse(_profile, _messages));
            Self.Tell(PoisonPill.Instance);
        }
    }

    private IActorRef _originalSender;
    private UserProfile _profile;
    private List<Message> _messages;
}

这只是创建了两个从属参与者，向他们发送消息以进行破解，然后等待两者都响应，然后再将收集到的所有数据发送回原始请求者。

那么，这似乎是一种合理的方式来协调两个不同的响应，以便将它们组合起来？有没有比自己制作更简单的方法？

提前感谢您的回复！

Answer 1

谢谢大家，所以我现在根据 Roger 和 Jeff 的建议将演员简化为以下内容...

public class TaskBasedUserLoader : ReceiveActor
{
    public TaskBasedUserLoader()
    {
        Receive<LoadUserRequest>(msg => LoadProfileAndMessages(msg));
    }

    private void LoadProfileAndMessages(LoadUserRequest msg)
    {
        var originalSender = Sender;
        var loadPreferences = this.LoadProfile(msg.UserId);
        var loadMessages = this.LoadMessages(msg.UserId);

        Task.WhenAll(loadPreferences, loadMessages)
            .ContinueWith(t => new UserLoadedResponse(loadPreferences.Result, loadMessages.Result), 
                TaskContinuationOptions.AttachedToParent & TaskContinuationOptions.ExecuteSynchronously)
            .PipeTo(originalSender);
    }

    private Task<UserProfile> LoadProfile(string userId)
    {
        return Task.FromResult(new UserProfile { UserId = userId });
    }

    private Task<List<Message>> LoadMessages(string userId)
    {
        return Task.FromResult(new List<Message>());
    }
}

LoadProfile 和 LoadMessages 方法最终将调用存储库来获取数据，但现在我有一个简洁的方法来做我想做的事。

再次感谢！

Answer 2

恕我直言，这是一个有效的过程，因为您分叉操作然后加入它。

顺便说一句，您可以使用this.Self.GracefulStop(new TimeSpan(1)); 而不是发送毒丸。

Answer 3

您可以结合使用 Ask、WhenAll 和 PipeTo：

var task1 = actor1.Ask<Result1>(request1);
var task2 = actor2.Ask<Result2>(request2);

Task.WhenAll(task1, task2)
    .ContinueWith(_ => new Result3(task1.Result, task2.Result))
    .PipeTo(Self);

...

Receive<Result3>(msg => { ... });