c# - 实现 AsyncCodeActivities(使用 C# async/await)

Question

很长一段时间以来，我都在使用以下模板编写自定义 AsyncCodeActivity 类:

public sealed class MyActivity : AsyncCodeActivity<T>
{
    protected override IAsyncResult BeginExecute(AsyncCodeActivityContext context, AsyncCallback callback, object state)
    {
        var task = new Task<T>(this.Execute, state, CancellationToken.None, TaskCreationOptions.AttachedToParent);
        task.ContinueWith(s => callback(s));
        task.Start();
        return task;
    }

    protected override T EndExecute(AsyncCodeActivityContext context, IAsyncResult result)
    {
        var task = result as Task<T>;
        if (task.Exception != null)
        {
            // Error handling. Rethrow? Cancel?
        }

        return task.Result;
    }

    private T Execute(object state)
    {
        // Logic here
        return default(T);
    }
}

我对此有一些疑问:

1. 哪种方法是处理异常的正确方法？重新 throw ？将上下文设置为已取消？
2. 是否有一种优雅的方式来使用现在可用的 async/await 语法来编写它？

谢谢

Answer 1

1) 你应该从你的 EndExecute 中重新抛出异常方法。

2) 我建议您创建自己的基类型。我写了一个叫做 AsyncTaskCodeActivity<T>下面:

public abstract class AsyncTaskCodeActivity<T> : AsyncCodeActivity<T>
{
    protected sealed override IAsyncResult BeginExecute(AsyncCodeActivityContext context, AsyncCallback callback, object state)
    {
        var task = ExecuteAsync(context);
        var tcs = new TaskCompletionSource<T>(state);
        task.ContinueWith(t =>
        {
            if (t.IsFaulted)
                tcs.TrySetException(t.Exception.InnerExceptions);
            else if (t.IsCanceled)
                tcs.TrySetCanceled();
            else
                tcs.TrySetResult(t.Result);

            if (callback != null)
                callback(tcs.Task);
        });

        return tcs.Task;
    }

    protected sealed override T EndExecute(AsyncCodeActivityContext context, IAsyncResult result)
    {
        var task = (Task<T>)result;
        try
        {
            return task.Result;
        }
        catch (AggregateException ex)
        {
            ExceptionDispatchInfo.Capture(ex.InnerException).Throw();
            throw;
        }
    }

    protected abstract Task<T> ExecuteAsync(AsyncCodeActivityContext context);
}

如果你使用我的 AsyncEx库，这个包装器变得更加简单:

public abstract class AsyncTaskCodeActivity<T> : AsyncCodeActivity<T>
{
    protected sealed override IAsyncResult BeginExecute(AsyncCodeActivityContext context, AsyncCallback callback, object state)
    {
        var task = ExecuteAsync(context);
        return AsyncFactory<T>.ToBegin(task, callback, state);
    }

    protected sealed override T EndExecute(AsyncCodeActivityContext context, IAsyncResult result)
    {
        return AsyncFactory<T>.ToEnd(result);
    }

    protected abstract Task<T> ExecuteAsync(AsyncCodeActivityContext context);
}

一旦有了基类型，就可以定义自己的派生类型。这是一个使用 async 的/await :

public sealed class MyActivity : AsyncTaskCodeActivity<int>
{
    protected override async Task<int> ExecuteAsync(AsyncCodeActivityContext context)
    {
        await Task.Delay(100);
        return 13;
    }
}

这是将 CPU 密集型工作安排到线程池的一个(类似于您当前的模板):

public sealed class MyCpuActivity : AsyncTaskCodeActivity<int>
{
    protected override Task<int> ExecuteAsync(AsyncCodeActivityContext context)
    {
        return Task.Run(() => 13);
    }
}

---

评论更新:这是一个使用取消的评论。我不是 100% 确定它是正确的，因为取消本身是异步的，而且 AsyncCodeActivity<T>.Cancel 的语义记录不足(即，Cancel 是否应该在取消状态下等待事件完成？在调用 Cancel 后成功完成事件是否可以接受？)。

public abstract class AsyncTaskCodeActivity<T> : AsyncCodeActivity<T>
{
    protected sealed override IAsyncResult BeginExecute(AsyncCodeActivityContext context, AsyncCallback callback, object state)
    {
        var cts = new CancellationTokenSource();
        context.UserState = cts;
        var task = ExecuteAsync(context, cts.Token);
        return AsyncFactory<T>.ToBegin(task, callback, state);
    }

    protected sealed override T EndExecute(AsyncCodeActivityContext context, IAsyncResult result)
    {
        try
        {
            return AsyncFactory<T>.ToEnd(result);
        }
        catch (OperationCanceledException)
        {
            if (context.IsCancellationRequested)
                context.MarkCanceled();
            else
                throw;
            return default(T); // or throw?
        }
    }

    protected override void Cancel(AsyncCodeActivityContext context)
    {
        var cts = (CancellationTokenSource)context.UserState;
        cts.Cancel();
    }

    protected abstract Task<T> ExecuteAsync(AsyncCodeActivityContext context, CancellationToken cancellationToken);
}