Rx 中的多选多

Question

我有一个像这样的界面：

interface IProcessor{
    IObservable<Item> Process(Item item);
}

我有一个工作人员数组：

IProcessor[] _workers = ....

我想通过所有工作人员传递一个项目：

var ret = Observable.Return(item);
for (var i = 0; i < _workers.Length; i++)
{
    int index = i;
    ret = ret
        .SelectMany(r => _workers[index].Process(r))
    ;
}
return ret;

我对它的外观不太满意 - 有没有更干净的方法？

Answer 1

这对我有用：

IObservable<Item> ret = _workers.Aggregate(
    Observable.Return(item),
    (rs, w) =>
        from r in rs
        from p in w.Process(r)
        select p);

请记住，这种可观察量的聚合 - 无论是在你的问题还是在我的答案中 - 都会很快导致内存问题（即堆栈溢出）。在我的测试中，我可以让 400 名工人工作，但 500 名工人会导致崩溃。

您最好将您的 IProcessor 更改为不使用可观察量并像这样实现您的可观察量：

interface IProcessor{
    Item Process(Item item);
}

var f =
    _workers.Aggregate<IProcessor, Func<Item, Item>>(
            i => i,
            (fs, p) => i => p.Process(fs(i)));

var ret = Observable.Start(() => f(item), Scheduler.ThreadPool);

通过这种方法，我可以在堆栈溢出之前获得超过 20,000 个嵌套工作线程，并且结果几乎是瞬时达到该级别的。

Answer 2

也许是这样的：？

var item = new Item();
_workers
  .ToObservable()
  .SelectMany(worker => worker.Process(item))
  .Subscribe(item => ...);

我假设工作人员可以并行处理该项目。

PS如果你想要顺序处理，那就是

var item = new Item();
_workers
  .ToObservable()
  .Select(worker => worker.Process(item))
  .Concat()
  .Subscribe(item => ...);