反应式扩展 Observable.FromAsync：如何等到异步操作完成答案

【问题标题】：Reactive Extensions Observerable.FromAsync: How to wait until async operation is finished反应式扩展 Observable.FromAsync：如何等到异步操作完成
【发布时间】：2020-06-16 07:15:04
【问题描述】：

在我的应用程序中，我从消息总线（rabbit mq，但实际上并不重要）获取事件。此事件的处理需要相当长的时间，并且这些事件是突发产生的。一次只能处理一个事件。为了实现这一点，我使用 Rx 来序列化事件并异步执行处理，以免阻塞生产者。

这是（简化的）示例代码：

static void Main(string[] args)
{
   var input = Observable.Interval(TimeSpan.FromMilliseconds(500));
   var subscription = input.Select(i => Observable.FromAsync(ct => Process(ct, i)))
                           .Concat()
                           .Subscribe();

   Console.ReadLine();
   subscription.Dispose();
   // Wait until finished?
   Console.WriteLine("Completed");
}

private static async Task Process(CancellationToken cts, long i)
{
   Console.WriteLine($"Processing {i} ...");
   await Task.Delay(1000).ConfigureAwait(false);
   Console.WriteLine($"Finished processing {i}");
}

示例应用程序正在处理订阅，然后应用程序被终止。但在此示例中，应用程序正在终止，而订阅前收到的最后一个事件仍在处理中。

问题：在处理订阅后等待处理最后一个事件的最佳方法是什么？我仍在尝试围绕异步 Rx 的东西。我认为有一种相当简单的方法可以做到这一点，但我现在还没有看到。

【问题讨论】：

RX 有一个针对schedulers 的内置功能。您可以选择使用ObserveOn() 而不是使用异步吗？查看其他问题，例如stackoverflow.com/questions/42655958/…

标签： c# system.reactive

【解决方案1】：

如果您对适用于此类简单情况的快速简便的解决方案感到满意，那么您可以只使用Wait IObservable 而不是订阅它。如果您想接收类似订阅的通知，请在最后的 Wait 之前使用 Do 运算符：

static void Main(string[] args)
{
    var input = Observable.Interval(TimeSpan.FromMilliseconds(500));
    input.Select(i => Observable.FromAsync(ct => Process(ct, i)))
        .Concat()
        .Do(onNext: x => { }, onError: ex => { }, onCompleted: () => { })
        .Wait();
    Console.WriteLine("Completed");
}

【讨论】：

感谢您的建议。我可以使用快速简便的解决方案，但不能使用“等待”。但这你不能从我的问题中知道:)。但基本上它非常相似。再次感谢。这很有帮助，我当然会接受你的回答。
除了阻塞，你也可以直接await你的observables '等待'直到完成。

【解决方案2】：

感谢 Theodor 的建议，我现在找到了适合我的解决方案：

static async Task Main(string[] args)
{
    var inputSequence = Observable.Interval(TimeSpan.FromMilliseconds(500));
    var terminate = new Subject<Unit>();
    var task = Execute(inputSequence, terminate);

    Console.ReadLine();
    terminate.OnNext(Unit.Default);
    await task.ConfigureAwait(false);
    Console.WriteLine($"Completed");
    Console.ReadLine();
}

private static async Task Process(CancellationToken cts, long i)
{
    Console.WriteLine($"Processing {i} ...");
    await Task.Delay(1000).ConfigureAwait(false);
    Console.WriteLine($"Finished processing {i}");
}

private static async Task Execute(IObservable<long> input, IObservable<Unit> terminate)
{
    await input
          .TakeUntil(terminate)
          .Select(i => Observable.FromAsync(ct => Process(ct, i)))
          .Concat();
}

【讨论】：