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

Question

我有这个代码片段：

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 次，每次循环后都会打印一行

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

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

谢谢。

Answer 1

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

但它们不会阻止 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");
}

这完全符合您的预期。

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

Why C# Rx Subscribe() function doesn't work with "async" keyword?

c#

asynchronous

async-await

system.reactive

subscribe