异步并等待使用 Task 的同步方法

Question

我对这个 async/await 概念非常陌生，所以对于提出任何显而易见的问题，我深表歉意。

我需要发送电子邮件，而新的 API 要求我使用 async 并等待。问题是我的很多方法都需要同步调用这个“发送电子邮件”。

所以，我创建了一个同步包装方法：

private Task SendEmailAsync(string email, string content)
{
   ...
   ...
   ...
}

private void SendEmail(string email, string content)
{
    Task tsk = Task.Factory.StartNew(async () => await SendEmailAsync(email, content));
    try { tsk.Wait(); }
    catch (AggregateException aex)
    {
        throw aex.Flatten();
    }
}

但由于某种原因，tsk.Wait() 不会等待 await SendEmailAsync(...) 完成。所以，我需要添加ManualResetEvent。像这样的

private void SendEmail(string email, string content)
{
    ManualResetEvent mre = new ManualResetEvent(false);
    Task tsk = Task.Factory.StartNew(async () =>
    {
        mre.Reset();
        try { await SendEmailAsync(email, content); }
        finally { mre.Set(); }
    });

    mre.WaitOne();
    try { tsk.Wait(); }
    catch (AggregateException aex)
    {
        throw aex.Flatten();
    }
}

但是 SendEmailAsync(...) 抛出的任何异常都不会被 tsk.Wait() 捕获。我的问题是：

1. 为什么 tsk.Wait() 不等待 await SendEmailAsync(...)
2. 如何捕获 await SendEmailAsync(...) 引发的异常

谢谢！

Answer 1

您可以使用以下扩展以同步方式运行异步代码。

https://stackoverflow.com/a/5097066/5062791

public static class AsyncHelpers
{
    /// <summary>
    /// Execute's an async Task<T> method which has a void return value synchronously
    /// </summary>
    /// <param name="task">Task<T> method to execute</param>
    public static void RunSync(Func<Task> task)
    {
        var oldContext = SynchronizationContext.Current;
        var synch = new ExclusiveSynchronizationContext();
        SynchronizationContext.SetSynchronizationContext(synch);
        synch.Post(async _ =>
        {
            try
            {
                await task();
            }
            catch (Exception e)
            {
                synch.InnerException = e;
                throw;
            }
            finally
            {
                synch.EndMessageLoop();
            }
        }, null);
        synch.BeginMessageLoop();

        SynchronizationContext.SetSynchronizationContext(oldContext);
    }

    /// <summary>
    /// Execute's an async Task<T> method which has a T return type synchronously
    /// </summary>
    /// <typeparam name="T">Return Type</typeparam>
    /// <param name="task">Task<T> method to execute</param>
    /// <returns></returns>
    public static T RunSync<T>(Func<Task<T>> task)
    {
        var oldContext = SynchronizationContext.Current;
        var synch = new ExclusiveSynchronizationContext();
        SynchronizationContext.SetSynchronizationContext(synch);
        T ret = default(T);
        synch.Post(async _ =>
        {
            try
            {
                ret = await task();
            }
            catch (Exception e)
            {
                synch.InnerException = e;
                throw;
            }
            finally
            {
                synch.EndMessageLoop();
            }
        }, null);
        synch.BeginMessageLoop();
        SynchronizationContext.SetSynchronizationContext(oldContext);
        return ret;
    }

    private class ExclusiveSynchronizationContext : SynchronizationContext
    {
        private bool done;
        public Exception InnerException { get; set; }
        readonly AutoResetEvent workItemsWaiting = new AutoResetEvent(false);
        readonly Queue<Tuple<SendOrPostCallback, object>> items =
            new Queue<Tuple<SendOrPostCallback, object>>();

        public override void Send(SendOrPostCallback d, object state)
        {
            throw new NotSupportedException("We cannot send to our same thread");
        }

        public override void Post(SendOrPostCallback d, object state)
        {
            lock (items)
            {
                items.Enqueue(Tuple.Create(d, state));
            }
            workItemsWaiting.Set();
        }

        public void EndMessageLoop()
        {
            Post(_ => done = true, null);
        }

        public void BeginMessageLoop()
        {
            while (!done)
            {
                Tuple<SendOrPostCallback, object> task = null;
                lock (items)
                {
                    if (items.Count > 0)
                    {
                        task = items.Dequeue();
                    }
                }
                if (task != null)
                {
                    task.Item1(task.Item2);
                    if (InnerException != null) // the method threw an exeption
                    {
                        throw new AggregateException("AsyncHelpers.Run method threw an exception.", InnerException);
                    }
                }
                else
                {
                    workItemsWaiting.WaitOne();
                }
            }
        }

        public override SynchronizationContext CreateCopy()
        {
            return this;
        }
    }
}

Answer 2

我应该首先说明您应该遵循@ChrFin 的评论并尝试重构您的代码以使其异步，而不是尝试同步运行现有的异步库（您甚至会注意到应用程序性能的改进）。

---

要完成你所寻求的，你应该使用SendEmailAsync(email, content).GetAwaiter().GetResult()。

为什么不SendEmailAsync(email, content).Wait()？你可能会问。

嗯，有一个微妙的区别。 Wait() 会将任何运行时异常包装在 AggregateException 中，如果您试图捕获原始异常，这将使您的生活更加艰难。

GetAwaiter().GetResult() 只会抛出原始异常。

您的代码将如下所示：

private void SendEmail(string email, string content)
{
    try
    {
        SendEmailAsync(email, content).GetAwaiter().GetResult();
    }
    catch(Exception ex)
    {
        // ex is the original exception
        // Handle ex or rethrow or don't even catch
    }
}

Answer 3

我需要发送电子邮件，而新的 API 要求我使用 async 和 await。问题是我的很多方法都需要同步调用这个“发送电子邮件”。

最佳解决方案是删除“同步调用者”要求。相反，您应该允许 async 和 await 在您的代码库中自然增长。

由于某种原因，tsk.Wait() 没有等待 await SendEmailAsync(...) 完成。

那是因为您使用的是Task.Factory.StartNew，即dangerous API。

我曾尝试使用 Task.Result 和 Task.GetAwaiter().GetResult() 但它导致了死锁。

我详细解释了这个死锁on my blog。

我使用了 ColinM 发布的解决方案，效果很好。

这是一个危险的解决方案。我不建议您使用它，除非您完全了解它的工作原理。

如果您绝对必须实现同步调用异步代码的反模式，那么您需要使用 hack 来执行此操作。我在brownfield async 上的一篇文章中介绍了各种黑客攻击。在您的情况下，您可能只使用线程池 hack，这只是一行代码：

private void SendEmail(string email, string content) =>
    Task.Run(() => SendEmailAsync(email, content)).GetAwaiter().GetResult();

但是，正如我在此答案开头所述，理想 解决方案是只允许 async 增长。像这样的黑客攻击会限制您的 Web 服务器的可扩展性。