如何在 Rx 中使用异步方法订阅？答案

【问题标题】：How to Subscribe with async method in Rx?如何在 Rx 中使用异步方法订阅？
【发布时间】：2016-05-24 11:22:41
【问题描述】：

我有以下代码：

IObservable<Data> _source;

...

_source.Subscribe(StoreToDatabase);

private async Task StoreToDatabase(Data data) {
    await dbstuff(data);
}

但是，这不会编译。有什么方法可以异步观察数据吗？我试过async void，它有效，但我觉得给定的解决方案不可行。

我还检查了Reactive Extensions Subscribe calling await，但它没有回答我的问题（我不关心SelectMany 的结果。）

【问题讨论】：

Reactive Extensions Subscribe calling await的可能重复

标签： c# async-await system.reactive

【解决方案1】：

您不必关心SelectMany 结果。答案仍然是一样的......虽然你需要你的任务有一个返回类型（即Task<T>，而不是Task）。

Unit本质上等同于void，所以你可以使用它：

_source.SelectMany(StoreToDatabase).Subscribe();

private async Task<Unit> StoreToDatabase(Data data)
{
    await dbstuff(data);
    return Unit.Default;
}

这个SelectMany 重载接受Func<TSource, Task<TResult>，这意味着在任务完成之前，结果序列不会完成。

【讨论】：

但至少要在您的订阅方法中放置一个OnError 处理程序。如果您不这样做，并且_source 错误或StoreToDatabase 抛出，您的应用程序将会崩溃，或者最终处于未知状态。
我们可以摆脱Unit吗？

【解决方案2】：

迟到的答案，但我认为以下扩展方法正确地封装了 Charles Mager 在his answer 中提出的内容：

public static IDisposable SubscribeAsync<T>(this IObservable<T> source, 
                         Func<Task> asyncAction, Action<Exception> handler = null)
{
    Func<T,Task<Unit>> wrapped = async t =>
    {
        await asyncAction();
        return Unit.Default;
    };
    if(handler == null)
        return source.SelectMany(wrapped).Subscribe(_ => { });
    else
        return source.SelectMany(wrapped).Subscribe(_ => { }, handler);
}

public static IDisposable SubscribeAsync<T>(this IObservable<T> source, 
                         Func<T,Task> asyncAction, Action<Exception> handler = null)
{
    Func<T, Task<Unit>> wrapped = async t =>
    {
        await asyncAction(t);
        return Unit.Default;
    };
    if(handler == null)
        return source.SelectMany(wrapped).Subscribe(_ => { });
    else
        return source.SelectMany(wrapped).Subscribe(_ => { }, handler);
}

【讨论】：

【解决方案3】：

我一直在使用 TPL DataFlow 来控制背压，并用它来解决这个问题。

关键部分是ITargetBlock<TInput>.AsObserver() - source。

// Set a block to handle each element
ITargetBlock<long> targetBlock = new ActionBlock<long>(async p =>
{
    Console.WriteLine($"Received {p}");
    await Task.Delay(1000);
    Console.WriteLine($"Finished handling {p}");
},
new ExecutionDataflowBlockOptions { BoundedCapacity = 1 });

// Generate an item each second for 10 seconds
var sequence = Observable.Interval(TimeSpan.FromSeconds(1)).Take(10);

// Subscribe with an observer created from the target block.
sequence.Subscribe(targetBlock.AsObserver());

// Await completion of the block
await targetBlock.Completion;

这里的重要部分是将 ActionBlock 的有界容量设置为 1。这可以防止块一次接收多个项目，如果项目已被处理，则 will block OnNext！

我最大的惊喜是在订阅中调用Task.Wait 和Task.Result 是安全的。显然，如果您调用了ObserverOnDispatcher() 或类似名称，您可能会遇到死锁。小心！

【讨论】：

【解决方案4】：

因此，您希望运行存储数据过程，可能还有其他一些过程，并异步等待完成或部分结果。此处显示的创建构造函数怎么样：

IObservable<Int32> NotifyOfStoringProgress = 
Observable.Create(
(Func<IObserver<Int32>, Task>)
(async (ObserverToFeed) =>
{
    ObserverToFeed.OnNext(-1);    
    Task StoreToDataBase = Task.Run(()=> {;});
    ObserverToFeed.OnNext(0);    
    ;;
    await StoreToDataBase;
    ObserverToFeed.OnNext(1);
    ;;
}));

NotifyOfStoringProgress.Subscribe(onNext: Notification => {;});

【讨论】：