system.reactive - RX で複数のサブスクリプションを同期する

Question

異なるオブザーバブルへの複数の RX サブスクリプションを強制的に (同時にではなく) 連続して実行することはできますか?

そのために EventLoopScheduler を使用できることは認識していますが、すべての処理が単一のスレッドで行われるため、パフォーマンスが低下します。

Answer 1

1 つのオブザーバブルを実行してからOnCompleted次のオブザーバブルを開始する場合は、Concat. 同時にサブスクライブする複数の異なるオブザーバブルを使用するMerge場合は、(セマンティクスがシナリオにとって意味がある場合) を使用できます。Merge が適切でない場合は、observer メソッドまたは既に知っている EventLoopScheduler で、標準のスレッド同期メソッド (lock、Monitor など) のいずれかを使用することをお勧めします。

EDIT元の回答を以下に保存

はい、シリアル オブザーバーの実行を強制することは可能です。ただし、必要かどうかはオブザーバブルによって異なります。一般に、ホット オブザーバブルはすでに連続して実行されますが、コールド オブザーバブルはそうではありません。これは、ホット オブザーバブルとコールド オブザーバブルの動作の違いによる副作用です。コールドのオブザーバブルをホットにして、オブザーバーを連続して実行するには、 を使用しますPublish。これは、さまざまな動作を示す例です。

Sub Main()
    'hot observable, runs serially
    Dim trigger As New ObsEvent
    Dim eobs = Observable.FromEventPattern(Of ItemEventArgs(Of String))(
                    Sub(h) AddHandler trigger.Triggered, h,
                    Sub(h) RemoveHandler trigger.Triggered, h)
    Dim sub1 = eobs.Subscribe(Sub(v)
                                  Console.WriteLine("Starting event observer 1: {0}", v.EventArgs.Item)
                                  Thread.Sleep(2000)
                                  Console.WriteLine("Ending event observer 1")
                              End Sub)
    trigger.Trigger("event trigger 1")
    Dim sub2 = eobs.Subscribe(Sub(v)
                                  Console.WriteLine("Starting event observer 2: {0}", v.EventArgs.Item)
                                  Thread.Sleep(2000)
                                  Console.WriteLine("Ending event observer 2")
                              End Sub)
    trigger.Trigger("event trigger 2")

    Console.WriteLine()
    Console.WriteLine()

    'cold observable, runs "simultaneously"
    Dim tobs = Observable.Timer(TimeSpan.FromSeconds(5))
    sub1 = tobs.Subscribe(Sub(v)
                              Console.WriteLine("Starting timer observer 1")
                              Thread.Sleep(2000)
                              Console.WriteLine("Ending timer observer 1")
                          End Sub,
                          Sub(ex) Console.WriteLine("Error"),
                          Sub() Console.WriteLine("Observer 1 completed"))
    Thread.Sleep(500)
    sub2 = tobs.Subscribe(Sub(v)
                              Console.WriteLine("Starting timer observer 2")
                              Thread.Sleep(2000)
                              Console.WriteLine("Ending timer observer 2")
                          End Sub,
                          Sub(ex) Console.WriteLine("Error"),
                          Sub() Console.WriteLine("Observer 2 completed"))

    'cold observable turned hot, runs serially
    Dim pobs = tobs.Publish()
    sub1 = pobs.Subscribe(Sub(v)
                              Console.WriteLine("Starting publish observer 1")
                              Thread.Sleep(2000)
                              Console.WriteLine("Ending publish observer 1")
                          End Sub,
                          Sub(ex) Console.WriteLine("Error"),
                          Sub() Console.WriteLine("Observer P1 completed"))
    Thread.Sleep(500)
    sub2 = pobs.Subscribe(Sub(v)
                              Console.WriteLine("Starting publish observer 2")
                              Thread.Sleep(2000)
                              Console.WriteLine("Ending publish observer 2")
                          End Sub,
                          Sub(ex) Console.WriteLine("Error"),
                          Sub() Console.WriteLine("Observer P2 completed"))
    pobs.Connect()

    Console.ReadKey()
End Sub