multithreading - Reactive Extensions 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)
);

Answer 2

呼び出しsomeSubject.OnNext()はスレッドセーフsomeList.Add()です-1 つ以上のスレッドから呼び出すことができますが、同時に呼び出すことはできません。OnNextステートメントでラップすれlockば安全です。

Answer 3

いいえ、シーケンスはシーケンシャルであることを意図しているため、重複する通知は許可されません。Synchronize 拡張メソッドを使用して、適切な同期を適用できます。Merge などのオペレーターは、On* コールバックでの適切なシリアル呼び出しを保証するために、ロックを取得してダウンストリーム オブザーバーを呼び出します。