c# - Observable 中的通知并行性

Question

我有以下可观察的

IObservable<Work> observable = SomeMethodToGetWorkToDo();

每次在上面调用 OnNext 时，我都需要在单独的线程中完成工作。每项工作都需要很长时间才能完成，所以我不能让队列中的其他 Work 项目等待太久，因为系统资源足够。

我以为 ObserveOn 可以解决我的问题，但是当我运行一些 Console.WriteLine 调用来查看线程 ID 时，我看到每个通知都有相同的线程 ID称呼。

如何确保 OnNext 中的并行性？

Answer 1

您需要将“要做的工作”转换为一系列“正在完成的工作”标记，然后您可以收集这些标记。最简单的方法是使用 TPL 将工作作为任务执行(最终将在线程池上运行任务)。像这样:

observable
    .SelectMany(work => Task.Run(() => DoWork(work))
    .Subscribe(workResult => Console.WriteLine("a work item was completed"));