将 IObservable<Task<T>> 解包为 IObservable 并保留<T>订单

Question

有没有办法将IObservable<Task<T>>解开为IObservable<T>保持相同的事件顺序，就像这样？

Tasks:  ----a-------b--c----------d------e---f---->
Values: -------A-----------B--C------D-----E---F-->

假设我有一个使用消息流的桌面应用程序，其中一些消息需要大量的后处理：

IObservable<Message> streamOfMessages = ...;
IObservable<Task<Result>> streamOfTasks = streamOfMessages
.Select(async msg => await PostprocessAsync(msg));
IObservable<Result> streamOfResults = ???; // unwrap streamOfTasks

我想有两种处理方式。

首先，我可以使用异步事件处理程序订阅streamOfTasks：

streamOfTasks.Subscribe(async task =>
{
var result = await task;
Display(result);
});

其次，我可以使用Observable.Create转换streamOfTasks，如下所示：

var streamOfResults =
from task in streamOfTasks
from value in Observable.Create<T>(async (obs, cancel) =>
{
var v = await task;
obs.OnNext(v);
// TODO: don't know when to call obs.OnComplete()
})
select value;
streamOfResults.Subscribe(result => Display(result));

无论哪种方式，消息的顺序都不会保留：一些后来的消息 不需要任何后处理比早期消息更快地出现 需要后处理。我的两个解决方案都处理传入的消息 并行处理，但我希望它们按顺序逐个处理。

我可以编写一个简单的任务队列来一次只处理一个任务， 但也许这是矫枉过正。在我看来，我错过了一些明显的东西。

---

上。我编写了一个示例控制台程序来演示我的方法。到目前为止，所有解决方案都不能保留事件的原始顺序。这是程序的输出：

Timer: 0
Timer: 1
Async handler: 1
Observable.Create: 1
Observable.FromAsync: 1
Timer: 2
Async handler: 2
Observable.Create: 2
Observable.FromAsync: 2
Observable.Create: 0
Async handler: 0
Observable.FromAsync: 0

以下是完整的源代码：

// "C:Program Files (x86)MSBuild14.0Bincsc.exe" test.cs /r:System.Reactive.Core.dll /r:System.Reactive.Linq.dll /r:System.Reactive.Interfaces.dll
using System;
using System.Reactive;
using System.Reactive.Concurrency;
using System.Reactive.Linq;
using System.Threading.Tasks;
class Program
{
static void Main()
{
Console.WriteLine("Press ENTER to exit.");
// the source stream
var timerEvents = Observable.Timer(TimeSpan.Zero, TimeSpan.FromSeconds(1));
timerEvents.Subscribe(x => Console.WriteLine($"Timer: {x}"));
// solution #1: using async event handler
timerEvents.Subscribe(async x =>
{
var result = await PostprocessAsync(x);
Console.WriteLine($"Async handler: {x}");
});
// solution #2: using Observable.Create
var processedEventsV2 =
from task in timerEvents.Select(async x => await PostprocessAsync(x))
from value in Observable.Create<long>(async (obs, cancel) =>
{
var v = await task;
obs.OnNext(v);
})
select value;
processedEventsV2.Subscribe(x => Console.WriteLine($"Observable.Create: {x}"));
// solution #3: using FromAsync, as answered by @Enigmativity
var processedEventsV3 =
from msg in timerEvents
from result in Observable.FromAsync(() => PostprocessAsync(msg))
select result;
processedEventsV3.Subscribe(x => Console.WriteLine($"Observable.FromAsync: {x}"));
Console.ReadLine();
}
static async Task<long> PostprocessAsync(long x)
{
// some messages require long post-processing
if (x % 3 == 0)
{
await Task.Delay(TimeSpan.FromSeconds(2.5));
}
// and some don't
return x;
}
}

Answer 1

将@Enigmativity的简单方法与@VMAtm附加计数器的想法以及这个 SO 问题中的一些代码片段相结合，我想出了这个解决方案：

// usage
var processedStream = timerEvents.SelectAsync(async t => await PostprocessAsync(t));
processedStream.Subscribe(x => Console.WriteLine($"Processed: {x}"));
// my sample console program prints the events ordered properly:
Timer: 0
Timer: 1
Timer: 2
Processed: 0
Processed: 1
Processed: 2
Timer: 3
Timer: 4
Timer: 5
Processed: 3
Processed: 4
Processed: 5
....

这是我SelectAsync扩展方法，用于将IObservable<Task<TSource>>转换为保持事件的原始顺序IObservable<TResult>：

public static IObservable<TResult> SelectAsync<TSource, TResult>(
this IObservable<TSource> src,
Func<TSource, Task<TResult>> selectorAsync)
{
// using local variable for counter is easier than src.Scan(...)
var counter = 0;
var streamOfTasks =
from source in src
from result in Observable.FromAsync(async () => new
{
Index = Interlocked.Increment(ref counter) - 1,
Result = await selectorAsync(source)
})
select result;
// buffer the results coming out of order
return Observable.Create<TResult>(observer =>
{
var index = 0;
var buffer = new Dictionary<int, TResult>();
return streamOfTasks.Subscribe(item =>
{
buffer.Add(item.Index, item.Result);
TResult result;
while (buffer.TryGetValue(index, out result))
{
buffer.Remove(index);
observer.OnNext(result);
index++;
}
});
});
}

我对我的解决方案不是特别满意，因为它对我来说看起来太复杂了，但至少它不需要任何外部依赖项。我在这里使用一个简单的字典来缓冲和重新排序任务结果，因为订阅者不需要是线程安全的(订阅是并发调用的)。

欢迎任何意见或建议。我仍然希望在没有自定义缓冲扩展方法的情况下找到本机 RX 方法来执行此操作。

Answer 2

RX 库包含三个运算符，可以解开可观察的任务序列，即Concat、Merge和Switch。这三者都接受一个IObservable<Task<T>>类型的source参数，并返回一个IObservable<T>。以下是文档中的描述：

Concat

连接所有任务结果，只要上一个任务成功终止。

Merge

将所有源任务的结果合并到单个可观察序列中。

Switch

将可观察

的任务序列转换为仅从最近的可观察序列生成值的可观察序列。每次收到新任务时，都会忽略前一个任务的结果。

换句话说，Concat按其原始顺序返回结果，Merge按完成顺序返回结果，Switch筛选出在发出下一个任务之前未完成的任务的任何结果。因此，只需使用内置的Concat运算符即可解决您的问题。不需要自定义运算符。

var streamOfResults = streamOfTasks
.Select(async task =>
{
var result1 = await task;
var result2 = await PostprocessAsync(result1);
return result2;
})
.Concat();

任务在由streamOfTasks发出之前就已启动。换句话说，它们正在以"热"状态出现。因此，Concat运算符一个接一个地等待它们这一事实对操作的并发性没有影响。它只影响其结果的顺序。如果您拥有冷可观察量(例如由Observable.FromAsync和Observable.Create方法创建的冷可观察量)，而不是热任务，在这种情况下，Concat将按顺序执行操作，这将是一个考虑因素。

Answer 3

以下简单的方法适合您吗？

IObservable<Result> streamOfResults =
from msg in streamOfMessages
from result in Observable.FromAsync(() => PostprocessAsync(msg))
select result;

Answer 4

要保持事件的顺序，您可以将流从 TPL 数据流汇集到TransformBlock中。默认情况下，TransformBlock将执行您的后处理逻辑，并将保持其输出的顺序。

using System;
using System.Collections.Generic;
using System.Reactive.Linq;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
using NUnit.Framework;
namespace HandlingStreamInOrder {
[TestFixture]
public class ItemHandlerTests {
[Test]
public async Task Items_Are_Output_In_The_Same_Order_As_They_Are_Input() {
var itemHandler = new ItemHandler();
var timerEvents = Observable.Timer(TimeSpan.Zero, TimeSpan.FromMilliseconds(250));
timerEvents.Subscribe(async x => {
var data = (int)x;
Console.WriteLine($"Value Produced: {x}");                
var dataAccepted = await itemHandler.SendAsync((int)data);
if (dataAccepted) {
InputItems.Add(data);
}                
});
await Task.Delay(5000);
itemHandler.Complete();
await itemHandler.Completion;
CollectionAssert.AreEqual(InputItems, itemHandler.OutputValues);
}
private IList<int> InputItems {
get;
} = new List<int>();
}
public class ItemHandler {

public ItemHandler() {            
var options = new ExecutionDataflowBlockOptions() {
BoundedCapacity = DataflowBlockOptions.Unbounded,
MaxDegreeOfParallelism = Environment.ProcessorCount,
EnsureOrdered = true
};
PostProcessBlock = new TransformBlock<int, int>((Func<int, Task<int>>)PostProcess, options);
var output = PostProcessBlock.AsObservable().Subscribe(x => {
Console.WriteLine($"Value Output: {x}");
OutputValues.Add(x);
});
}
public async Task<bool> SendAsync(int data) {
return await PostProcessBlock.SendAsync(data);
}
public void Complete() {
PostProcessBlock.Complete();
}
public Task Completion {
get { return PostProcessBlock.Completion; }
}
public IList<int> OutputValues {
get;
} = new List<int>();
private IPropagatorBlock<int, int> PostProcessBlock {
get;
}
private async Task<int> PostProcess(int data) {
if (data % 3 == 0) {
await Task.Delay(TimeSpan.FromSeconds(2));
}            
return data;
}
}
}

Answer 5

>Rx和TPL可以很容易地在这里组合，默认情况下，TPL保存事件的顺序，所以你的代码可能是这样的：

using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
static async Task<long> PostprocessAsync(long x) { ... }
IObservable<Message> streamOfMessages = ...;
var streamOfTasks = new TransformBlock<long, long>(async msg => 
await PostprocessAsync(msg)
// set the concurrency level for messages to handle
, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = Environment.ProcessorCount });
// easily convert block into observable
IObservable<long> streamOfResults = streamOfTasks.AsObservable();

编辑：Rx扩展旨在成为 UI 事件的反应式管道。由于这种类型的应用程序通常是单线程的，因此消息是通过保存顺序来处理的。但在一般情况下，C#中的事件不是线程安全的，因此您必须提供一些额外的逻辑来相同的顺序。

如果你不喜欢引入另一个依赖项的想法，你需要将操作编号存储在Interlocked类中，如下所示：

// counter for operations get started
int operationNumber = 0;
// counter for operations get done
int doneNumber = 0;
...
var currentOperationNumber = Interlocked.Increment(ref operationNumber);
...
while (Interlocked.CompareExchange(ref doneNumber, currentOperationNumber + 1, currentOperationNumber) != currentOperationNumber)
{
// spin once here
}
// handle event
Interlocked.Increment(ref doneNumber);