ここでSubscribeOn（）は何をしていますか？

Question

私はランダムな問題を解決し、愚かな初心者の質問に恥ずかしがりをすることで、自分のやり方を変えて反応的なプログラミングを教えています。スレッドスケジューリングがどのように機能するかを理解しながら、私は自分自身を切り刻むことができました。私はこのコードが理にかなっていないと確信していますが、何が起こっているのか理解できません。それがおそらく私を助けてくれるだろうとわかる。ここでは、コードです：

var testScheduler = new TestScheduler(); 
var newThreadScheduler = new NewThreadScheduler(); 

var emitter = new Subject<string>(); 
testScheduler.Schedule(TimeSpan.FromSeconds(0.1),() => emitter.OnNext("one")); 
testScheduler.Schedule(TimeSpan.FromSeconds(0.2),() => emitter.OnCompleted()); 

var subscription = emitter.SubscribeOn(newThreadScheduler) 
          .Subscribe(
           item => Console.WriteLine(item), 
           error => Console.WriteLine(error), 
           () => Console.WriteLine("Complete!") 
          ); 

testScheduler.AdvanceBy(TimeSpan.FromSeconds(1).Ticks); 

Console.WriteLine("DONE."); 
Console.ReadLine(); 

I 期待は多分何だった：可能なインタリーブで

one  
DONE. 
Complete! 

、私は何SubscribeOn（）はどうなるかなり確実ではなかったよう。私が代わりに得たことだった。まさにここで何が起こっている

DONE. 
Complete! 

？完成前に商品が生産されなかったのはなぜですか？ ObserveOn（）は私がこのケースで期待したとおりに動作し、理由を理解しています。デリゲートを他のスレッドで実行していて、「完了」でインターリーブできます。では、SubscribeOn（）は何をしているのですか？

Answer 1

ここにあるものは、単に競合状態です。我々はすべてのコードをバックリッピング場合

はちょうど

var emitter = new Subject<string>(); 
emitter.OnNext("one"); 
emitter.OnCompleted(); 

var subscription = emitter 
          .Subscribe(
           item => Console.WriteLine(item), 
           error => Console.WriteLine(error), 
           () => Console.WriteLine("Complete!") 
          ); 



Console.WriteLine("DONE."); 
Console.ReadLine(); 

私たちは、同じ結果を取得します。 Subject<T>を使用すると、キャッシング動作は発生しませんが、OnCompleted通知は例外です。

SubscribeOnオペレータは、提供されたISchedulerインスタンスで行われるすべてのサブスクリプション作業をスケジュールします。 Subject<T>に加入している場合は、ほとんど行われていません。 コールバックのリストにコールバックを登録するのと同じくらい簡単です。

NewThreadSchedulerにスケジュールを設定すると、新しいスレッドが作成され、スケジュールされた作業を処理する内部イベントループが作成されます。 これはかなり速いですが、新しいスレッド、EventloopSchedulerを作成し、新しいスレッドへのコンテキスト切り替えを実行する必要があります。

この例では、TestSchedulerにOnNextとOnCompletedの通知をスケジュールしています。 SubscribeOnの場合はNewThreadSchedulerです。 それに続いて、TestSchedulerインスタンスのすべてのスケジュールされた作業を処理し始めます。 の処理は実質的にです。予定されている項目を繰り返して、代理人を出して仮想時計を進めるだけです。 これは信じられないほど速いです。より具体的には

、以下のコードは、ここでは単純に（契約者またはオブザーバーそれらを呼び出す）コールバックアクションのリストを持っているあなたは

var newThreadScheduler = new NewThreadScheduler(); 

var callbacks = new List<Action<string>>(); 
newThreadScheduler.Schedule(()=>callbacks.Add(str=>Console.WriteLine(str))); 

foreach (var callback in callbacks) 
{ 
    callback("one"); 
} 

Console.WriteLine("Done"); 

を書かれているものに似ています。 新しいスレッドで、これらのコールバックの1つの追加を非同期にスケジュールします。 そして、直ちにコールバックを反復し、文字列 "one"をそれぞれに送ります。メインスレッドがコレクションを反復処理することができる前に は結果が

Done 

NewThreadSchedulerはちょうど、新しいスレッドを起動し、アクションをスケジュールし、そのアクションを実行するのに十分な時間を与えられていませんさ。

したがって、私があなたに従わなかったと考える2つのガイドラインがあります： 1）被験者;-) 2）スレッドとユニットテストを混在させないでください。これをテストしているので、TestSchedulerの存在が想定されます。ただし、TestSchedulerという2つのインスタンスを使用できます。背景と前景のインスタンス。

もっと役立つように、テストから2番目のスケジューラを削除することを提案する肯定的な指針を提供します。 SubscribeOn演算子のTestSchedulerインスタンスを使用してください。

次は、TestSchedulerの観測可能なシーケンスファクトリメソッド（CreateColdObservable）を使用して、件名+スケジューリングの使用を置き換えることをお勧めします。 最後に、Startメソッドを使用するだけで、speicifc時間が1秒になると何かが得られるかどうかはわかりません。 私はそれが魔法の値1の騒音と使用を減らすと思います。

var testScheduler = new TestScheduler(); 

var source = testScheduler.CreateColdObservable<string>(
    ReactiveTest.OnNext(TimeSpan.FromSeconds(0.1).Ticks, "one"), 
    ReactiveTest.OnCompleted<string>(TimeSpan.FromSeconds(0.2).Ticks)); 

var subscription = source.SubscribeOn(testScheduler) 
          .Subscribe(
           item => Console.WriteLine(item), 
           error => Console.WriteLine(error), 
           () => Console.WriteLine("Complete!") 
          ); 

testScheduler.Start(); 

Console.WriteLine("DONE."); 
Console.ReadLine(); 

唯一の問題は、SubscribeOnコールがかなり冗長であることです。

FYI：NewThreadSchedulerのコード - 私は反応性の何も知らないが、あなたのコードを見て https://github.com/Reactive-Extensions/Rx.NET/blob/master/Rx.NET/Source/System.Reactive.PlatformServices/Reactive/Concurrency/NewThreadScheduler.cs