c# - 使用 SubscribeOn(Scheduler.TaskPool) 时避免在 Rx 中重叠 OnNext 调用

Question

我有一些使用 Rx 的代码，从多个线程调用:

subject.OnNext(value); // where subject is Subject<T>

我希望在后台处理这些值，所以我的订阅是

subscription = subject.ObserveOn(Scheduler.TaskPool).Subscribe(value =>
{
    // use value
});

我真的不关心哪些线程处理来自 Observable 的值，只要工作被放入 TaskPool 并且不阻塞当前线程即可。但是，我在 OnNext 委托(delegate)中使用的“值”不是线程安全的。目前，如果有很多值通过 Observable，我会收到对 OnNext 处理程序的重叠调用。

我可以只向我的 OnNext 委托(delegate)添加一个锁，但这不像 Rx 做事的方式。当我有多个线程调用 subject.OnNext(value); 时，确保一次只调用 OnNext 处理程序的最佳方法是什么？

Answer 1

来自 Using Subjects在 MSDN 上

By default, subjects do not perform any synchronization across threads. [...] If, however, you want to synchronize outgoing calls to observers using a scheduler, you can use the Synchronize method to do so.

因此，正如 Brandon 在评论中所说，您应该同步主题并将其交给您的生产者线程。例如

var syncSubject = Subject.Synchronize(subject);

// syncSubject.OnNext(value) can be used from multiple threads

subscription = syncSubject.ObserveOn(TaskPoolScheduler.Default).Subscribe(value =>
{
    // use value
});