为什么 C# Rx Subscribe() 函数不适用于“async”关键字？

【问题标题】：Why C# Rx Subscribe() function doesn't work with "async" keyword?为什么 C# Rx Subscribe() 函数不适用于“async”关键字？
【发布时间】：2020-03-07 12:02:46
【问题描述】：

我有这个代码 sn-p:

static void Main(string[] args)
{
    Observable.Range(1, 5).Subscribe(async x => await DoTheThing(x));
    Console.WriteLine("done");
}

static async Task DoTheThing(int x)
{
    await Task.Delay(TimeSpan.FromSeconds(x));
    Console.WriteLine(x);
}

我希望它会循环 5 次，每次循环后都会打印一行

但令人惊讶的是，这将打印“完成”并立即终止。似乎 async+await 没有等待 Task.Delay 并退出。

语义似乎没有问题，那么我在订阅或异步方面哪里出错了，如何解决它以满足我从 Rx 调用异步任务的请求？

谢谢。

【问题讨论】：

您希望在 5 秒还是 15 秒内打印 1 到 5？

标签： c# asynchronous async-await system.reactive subscribe

【解决方案1】：

它不是阻塞的，因为它很好 - 异步。您的代码生成五个任务，全部并行运行，在不同时间完成。

但它们不会阻止Main 函数。如果您只是添加Console.ReadKey() 作为最后一行，您会看到您的代码确实在后台运行。它会打印出来。

static void Main(string[] args)
{
    Observable.Range(1, 5).Subscribe(async x => await DoTheThing(x));
    Console.WriteLine("done");
    Console.ReadKey();
}

但假设您想等到所有这些都完成。然后呢？

当然，有.Wait()，但那是阻塞的。让我们将所有任务作为可观察对象来观察。

我们将使用 C# 7 的 async Main。

static async Task Main(string[] args)
{
    await Observable.Range(1, 5)
        .Select(x => DoTheThing(x).ToObservable())
        .Merge();                

    Console.WriteLine("done");
}

这完全符合您的预期。

【讨论】：