如何将其转换为异步任务？

Question

给定以下代码...

static void DoSomething(int id) {
    Thread.Sleep(50);
    Console.WriteLine(@"DidSomething({0})", id);
}

我知道我可以将其转换为异步任务，如下所示...

static async Task DoSomethingAsync(int id) {
    await Task.Delay(50);
    Console.WriteLine(@"DidSomethingAsync({0})", id);
}

如果我多次调用 (Task.WhenAll)，这样做会比使用 Parallel.Foreach 甚至在循环中调用更快、更高效。

但是，让我们假设 Task.Delay() 不存在，我实际上必须使用 Thread.Sleep();我知道实际上情况并非如此，但这是概念代码，延迟/睡眠所在的位置通常是没有异步选项（例如早期 EF）的 IO 操作。

我已经尝试了以下...

static async Task DoSomethingAsync2(int id) {
    await Task.Run(() => {
        Thread.Sleep(50);
        Console.WriteLine(@"DidSomethingAsync({0})", id);
    });
}

但是，尽管它运行时没有错误，但根据Lucien Wischik，这实际上是一种不好的做法，因为它只是从池中启动线程来完成每个任务（使用以下控制台应用程序也会更慢 - 如果你交换在 DoSomethingAsync 和 DoSomethingAsync2 调用之间，您可以看到完成时间的显着差异...

static void Main(string[] args) {
    MainAsync(args).Wait();
}

static async Task MainAsync(String[] args) {

    List<Task> tasks = new List<Task>();
    for (int i = 1; i <= 1000; i++)
        tasks.Add(DoSomethingAsync2(i)); // Can replace with any version
    await Task.WhenAll(tasks);

}

然后我尝试了以下...

static async Task DoSomethingAsync3(int id) {
    await new Task(() => {
        Thread.Sleep(50);
        Console.WriteLine(@"DidSomethingAsync({0})", id);
    });
}

用它代替原来的 DoSomethingAsync，测试永远不会完成，屏幕上什么也没有显示！

我还尝试了多种其他变体，它们要么无法编译，要么无法完成！

因此，鉴于您不能调用任何现有的异步方法并且必须在异步任务中同时完成 Thread.Sleep 和 Console.WriteLine 的约束，您如何以与原始代码一样高效的方式完成它?

对于那些感兴趣的人来说，这里的目标是让我更好地了解如何创建我自己的异步方法，而我不调用任何其他人。尽管进行了多次搜索，但这似乎是真正缺乏示例的一个领域-尽管有成千上万个调用异步方法的示例依次调用其他异步方法，但我找不到任何将现有 void 方法转换为异步任务的示例除了使用 Task.Run(() => {} ) 方法的任务之外，没有对进一步异步任务的调用。

Answer 1

有两种任务：执行代码的任务（例如，Task.Run 和朋友），以及响应某些外部事件的任务（例如，TaskCompletionSource<T> 和朋友）。

您正在寻找的是TaskCompletionSource<T>。常见情况有多种“速记”形式，因此您不必总是直接使用TaskCompletionSource<T>。例如，Task.FromResult 或 TaskFactory.FromAsync。 FromAsync 是最常用的，如果你有一个现有的 *Begin/*End 实现你的 I/O；否则，您可以直接使用TaskCompletionSource<T>。

有关详细信息，请参阅Implementing the Task-based Asynchronous Pattern 的“I/O 绑定任务”部分。

Task 构造函数（不幸的是）是基于任务的并行性的保留，不应在异步代码中使用。只能用于创建基于代码的任务，不能用于创建外部事件任务。

因此，鉴于您不能调用任何现有的异步方法并且必须在异步任务中同时完成 Thread.Sleep 和 Console.WriteLine 的约束，您如何以与原始代码一样高效的方式完成它?

我会使用某种计时器并让它在计时器触发时完成TaskCompletionSource<T>。我几乎可以肯定这就是实际的Task.Delay 实现无论如何。

Answer 2

因此，鉴于您不能调用任何现有的约束 异步方法，并且必须同时完成 Thread.Sleep 和 Console.WriteLine 在一个异步任务中，你是怎么做到的 方式和原始代码一样高效？

IMO，这是一个非常合成约束，你真的需要坚持Thread.Sleep。在此约束下，您仍然可以稍微改进基于 Thread.Sleep 的代码。而不是这个：

static async Task DoSomethingAsync2(int id) {
    await Task.Run(() => {
        Thread.Sleep(50);
        Console.WriteLine(@"DidSomethingAsync({0})", id);
    });
}

你可以这样做：

static Task DoSomethingAsync2(int id) {
    return Task.Run(() => {
        Thread.Sleep(50);
        Console.WriteLine(@"DidSomethingAsync({0})", id);
    });
}

这样，您可以避免编译器生成的状态机类的开销。在how exceptions are propagated 中，这两个代码片段之间存在细微差别。

无论如何，这不是减速的瓶颈所在。

（使用以下控制台应用程序也会更慢 - 如果您 在 DoSomethingAsync 和 DoSomethingAsync2 调用之间交换，您可以看到一个 完成时间的显着差异）

让我们再看看你的主循环代码：

static async Task MainAsync(String[] args) {

    List<Task> tasks = new List<Task>();
    for (int i = 1; i <= 1000; i++)
        tasks.Add(DoSomethingAsync2(i)); // Can replace with any version
    await Task.WhenAll(tasks);

}

从技术上讲，它要求并行运行 1000 个任务，每个任务都应该在自己的线程上运行。在理想的宇宙中，您希望并行执行 1000 次 Thread.Sleep(50) 并在大约 50 毫秒内完成整个事情。

但是，TPL 的默认任务调度程序永远不会满足此请求，这是有充分理由的：线程是一种宝贵且昂贵的资源。此外，实际并发操作数受限于 CPU/内核数。所以实际上，在默认大小为ThreadPool 的情况下，我得到21 个池线程（在峰值） 并行服务此操作。这就是为什么DoSomethingAsync2 / Thread.Sleep 比DoSomethingAsync / Task.Delay 花费的时间要长得多。 DoSomethingAsync 不会阻塞池线程，它只会在超时完成时请求一个。因此，更多的DoSomethingAsync 任务实际上可以并行运行，而不是DoSomethingAsync2 那些。

测试（控制台应用）：

// https://stackoverflow.com/q/21800450/1768303

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;

namespace Console_21800450
{
    public class Program
    {
        static async Task DoSomethingAsync(int id)
        {
            await Task.Delay(50);
            UpdateMaxThreads();
            Console.WriteLine(@"DidSomethingAsync({0})", id);
        }

        static async Task DoSomethingAsync2(int id)
        {
            await Task.Run(() =>
            {
                Thread.Sleep(50);
                UpdateMaxThreads();
                Console.WriteLine(@"DidSomethingAsync2({0})", id);
            });
        }

        static async Task MainAsync(Func<int, Task> tester)
        {
            List<Task> tasks = new List<Task>();
            for (int i = 1; i <= 1000; i++)
                tasks.Add(tester(i)); // Can replace with any version
            await Task.WhenAll(tasks);
        }

        volatile static int s_maxThreads = 0;

        static void UpdateMaxThreads()
        {
            var threads = Process.GetCurrentProcess().Threads.Count;
            // not using locks for simplicity
            if (s_maxThreads < threads)
                s_maxThreads = threads;
        }

        static void TestAsync(Func<int, Task> tester)
        {
            s_maxThreads = 0;
            var stopwatch = new Stopwatch();
            stopwatch.Start();

            MainAsync(tester).Wait();

            Console.WriteLine(
                "time, ms: " + stopwatch.ElapsedMilliseconds +
                ", threads at peak: " + s_maxThreads);
        }

        static void Main()
        {
            Console.WriteLine("Press enter to test with Task.Delay ...");
            Console.ReadLine();
            TestAsync(DoSomethingAsync);
            Console.ReadLine();

            Console.WriteLine("Press enter to test with Thread.Sleep ...");
            Console.ReadLine();
            TestAsync(DoSomethingAsync2);
            Console.ReadLine();
        }

    }
}

输出：

按 Enter 以使用 Task.Delay 进行测试... ... 时间，毫秒：1077，峰值线程：13 按 Enter 以使用 Thread.Sleep 进行测试... ... 时间，毫秒：8684，峰值线程：21

是否可以改进基于Thread.Sleep 的DoSomethingAsync2 的时序图？我能想到的唯一方法是将TaskCreationOptions.LongRunning 与Task.Factory.StartNew 一起使用：

在任何实际应用中执行此操作之前，您应该三思而后行：

static async Task DoSomethingAsync2(int id)
{
    await Task.Factory.StartNew(() =>
    {
        Thread.Sleep(50);
        UpdateMaxThreads();
        Console.WriteLine(@"DidSomethingAsync2({0})", id);
    }, TaskCreationOptions.LongRunning | TaskCreationOptions.PreferFairness);
}

// ...

static void Main()
{
    Console.WriteLine("Press enter to test with Task.Delay ...");
    Console.ReadLine();
    TestAsync(DoSomethingAsync);
    Console.ReadLine();

    Console.WriteLine("Press enter to test with Thread.Sleep ...");
    Console.ReadLine();
    TestAsync(DoSomethingAsync2);
    Console.ReadLine();
}

输出：

按 Enter 以使用 Thread.Sleep 进行测试... ... 时间，毫秒：3600，峰值线程：163

时机越来越好，但代价高昂。此代码要求任务调度程序为每个新任务创建一个新线程。 不要期望该线程来自池：

Task.Factory.StartNew(() =>
{
    Thread.Sleep(1000);
    Console.WriteLine("Thread pool: " + 
        Thread.CurrentThread.IsThreadPoolThread); // false!
}, TaskCreationOptions.LongRunning).Wait();