12

如何比较以下两者?Rx 更强大吗?

反应式扩展:

var observable = Observable.Create<char>(async (observer, cancel) =>
{
    while (true)
    {
        string line = await sr.ReadLineAsync();
        if (line == null)
            break;
        observer.OnNext(line);
    }
});

observable.Subscribe(
    c => Console.WriteLine(c.ToString()),
    () => end.Dispose());

异步流:

public async void Run(string path)
{
    await foreach (var line in TestAsync())
    {
        Console.WriteLine(line);
    }
}

private async IAsyncEnumerable<string> TestAsync()
{
    while (true)
    {
        string line = await sr.ReadLineAsync();
        if (line == null)
            break;
        yield return line;
    }
}
4

1 回答 1

12

这两个功能一起工作。PS:算了async streams,想想吧await foreach

异步流

异步流是允许异步迭代的(相对)低级功能。它本身不提供任何其他功能,如过滤、聚合等。它是基于拉的,而 Rx 是基于推的。

您可以通过ReacticeX.NET Github repo中的System.Linq.Async库在异步流上使用 LINQ 运算符。它很快,但不提供 Rx 的事件处理功能。

例如,没有时间的概念,更不用说使用自定义调度程序的方法了。没有订阅,没有错误事件。GroupBy 将消耗整个源并将组项作为单独IAsyncEnumerable的实例发出,而 Rx 的 GroupBy 将为每个组发出单独的 Observables。

在问题的示例中, IAsyncEnumerable 很自然,因为不涉及事件逻辑,只是在异步迭代器上进行迭代。

如果该示例尝试轮询远程服务并检测故障峰值(即每个间隔的故障数超过阈值),则 IAsyncEnumerable 将不合适,因为它会阻止等待所有响应。事实上,我们根本无法聚合每个时间的事件。

穿线

真的没有——一个 IAsyncEnumerable 或await foreach调用不指定事件是如何产生或消费的。如果我们想使用单独的任务来处理一个项目,我们必须自己创建它,例如:

public async Task Run(string path)
{
    await foreach (var line in LoadStockTrades())
    {
        var result = await Task.Run(()=>AnalyzeTrade(line));
        Console.WriteLine($"{result} : {line});
    }
}

反应式扩展

Reactive Extensions 是一个处理事件流的高级库。它是基于推送的,它理解时间,但它也比异步流或通道等较低级别的构造慢。

在问题的示例中,Rx 将是矫枉过正。轮询和检测尖峰虽然很容易,但有多个窗口选项。

System.Linq.Async 可以使用ToObservable从 IAsyncEnumerable 创建 Observable ,这意味着 IAsyncEnumerable 可以用作 Rx 的源。

穿线

默认情况下,Rx 是单线程的,这对于它的主要场景 - 事件流处理非常有意义。

另一方面,Rx 允许发布者、订阅者和操作者在相同或不同的线程上运行。在没有DataFlow 或 DataFlow 的语言中async/await(例如 Java、JavaScript),Rx 用于通过在不同线程上运行发布者和订阅者来模拟并发处理管道。

于 2019-10-22T16:33:15.220 回答