在任务上使用akka.net PipeTo()进行异常处理

Question

参考Akka.Net文档,PipeTo()在处理异步作业时首选使用.

当处理返回的函数时Task<T>,我可以处理失败事件,没问题.

问题是,当处理一个不返回任何类型但只有Task一个PipeTo函数的函数时,仍然会调用该函数,但它不是包含失败句柄的重载,而是现在说明如下:'由于此任务没有结果,只有例外情况会通过管道传送给收件人.'

这是否意味着,如果我有以下代码:

public class RepositoryComponent : IRepositoryComponent
{
    private SqlSettings _sqlSettings;

    public RepositoryComponent(SqlSettings sqlSettings)
    {
        _sqlSettings = sqlSettings;
    }

    public async Task InsertJobAsync(RetryModel job)
    {
        try
        {
            await... //some logic
        }
        catch { throw; }
    }
}

我的演员:

public class RepositoryActor : ActorBase
{
    private IRepositoryComponent _repoComponent;
    private ActorSelection _retryActor;

    public RepositoryActor(IRepositoryComponent repoComponent) : base()
    {
        _repoComponent = repoComponent;
    }

    public override void Listening()
    {
        Receive<RepositoryMessages.GenericRequestNoReturn>(x => InvokeRequest(x));
        Receive<RepositoryMessages.GenericRequestWithResponseType>(x => InvokeRequestAndSendResponse(x));
    }

    private void InvokeRequest(RepositoryMessages.GenericRequestNoReturn msg)
    {
        try
        {
            //some logic with msg
            _repoComponent.InsertJobAsync(new Core.Models.RetryModel()).PipeTo(Context.Self);
        }
        catch (Exception ex)
        {
            base.ExceptionHandling(ex);
        }
    }
}

为了捕获上面actor中的异常,我需要添加另一个Receive处理程序来满足异常,例如,如下所示:

Receive<Exception>(x => SomeExceptionHandler(x));

它是否正确？如果是这样,try {} catch {}我的代码块周围不需要？

Answer 1

如果您正在调用异步操作PipeTo,则不会阻止当前的代码执行路径.这意味着永远不会执行try/catch语句(因为表示异步操作的任务的成功或失败将被重定向到Self).

有两种方法可以处理异步操作执行期间可能发生的异常:

1. 使用ReceiveAsync<>而不是Receive<>与异步lambda组合使用 - 这将允许您简单await地完成异步操作.您可以在此上下文中自由使用try/catch,因为它可以按预期工作.请记住,这将使您的actor不可重入 - 这意味着,在当前异步操作完成之前,actor不会处理任何消息.
2. 在逻辑中保留PipeTo并添加另一个Receive<>处理程序 - 如果包装异步操作的任务失败,异常将被包装到一个消息中并重定向回到PipeTo目标(Self在您的情况下).默认情况下,以这种方式生成的异常包含在Status.Failure消息中 - 在这种情况下,您需要Receive<Status.Failure>在actor中添加处理程序.您还可以指定用于处理PipeTo故障的自定义消息构造函数,即:_repoComponent.InsertJobAsync(input).PipeTo(Context.Self, failure: exception => new MyErrorMessage(exception))