リアクティブエクステンション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

いいえ、シーケンスは連続しているため、重複する通知は許可されません。同期メソッドを使用すると、適切な同期を実行できます。 Mergeのような演算子は、On *コールバックに対する適切なシリアル呼び出しを確実にするために、下流のオブザーバを呼び出すロックをとる。

Answer 3

someSubject.OnNext()を呼び出すのはsomeList.Add()と同じようにスレッドセーフです.1つ以上のスレッドから呼び出すことはできますが、同時に実行することはできません。あなたのOnNextをlockという声明で包みなさい。それは安全です。