multithreading - 响应式(Reactive)扩展 OnNext 线程安全

Question

与 Rx Subject , 调用 OnNext() 是线程安全的吗？从多个线程？

所以序列可以从多个来源生成。

合并会做同样的事情吗？

Answer 1

Rx 合约要求通知是顺序的，并且对于多个运营商来说是逻辑上的必要。也就是说，您可以使用可用的 Synchronize获得这种行为的方法。

var subject = new Subject<int>();
var syncedSubject = Subject.Synchronize(subject);            

您现在可以同时调用 syncedSubject .
对于必须同步的观察者，您还可以使用:

var observer = Observer.Create<Unit>(...);
var syncedObserver = Observer.Synchronize(observer);

测试:

Func<int, Action> onNext = i => () => syncedSubject.OnNext(i);
Parallel.Invoke
(
    onNext(1),
    onNext(2),
    onNext(3),
    onNext(4)
);