52286


C# の Reactive.Extensions について勉強中です。  
環境は、.NET Core 2.1, C# 7.2 です。

スケジューラについて、挙動の確認をしていたのですが、スケジューラを指定した時の動作が想定しているものと異なるため戸惑っています。

まず１つ目のソースとして、スケジューラを指定しないものが以下です。

[code]

    class Program
    {
        static void Main(string[] args)
        {
            Console.WriteLine($"Main Thread Id: {Thread.CurrentThread.ManagedThreadId}");
    
            var a = new A();
    
            a.Subscribe(onNext: x =>
            {
                Console.WriteLine($"Current Thread Id: {Thread.CurrentThread.ManagedThreadId}, Value:{x}");
            });
    
            a.Notify(1);
            a.Notify(2);
            a.Notify(3);
    
            Console.ReadKey();
        }
    }
    
    class A : IObservable<int>
    {
        readonly Subject<int> observers = new Subject<int>();
    
        public IDisposable Subscribe(IObserver<int> observer)
        {
            return observers.Subscribe(observer);
        }
    
        public void Notify(int x) => observers.OnNext(x);
    }
    
[/code]

これは、以下のように出力されます。

[code]

    Main Thread Id: 1
    Current Thread Id: 1, Value:1
    Current Thread Id: 1, Value:2
    Current Thread Id: 1, Value:3
    
[/code]

ここで、スケジューラとして `ThreadPoolScheduler` を以下のように指定しました。

[code]

    class Program
    {
        static void Main(string[] args)
        {
            Console.WriteLine($"Main Thread Id: {Thread.CurrentThread.ManagedThreadId}");
    
            var a = new A();
    
            // ↓ここです
            a.SubscribeOn(ThreadPoolScheduler.Instance).Subscribe(onNext: x =>
            {
                Console.WriteLine($"Current Thread Id: {Thread.CurrentThread.ManagedThreadId}, Value:{x}");
            });
    
            a.Notify(1);
            a.Notify(2);
            a.Notify(3);
    
            Console.ReadKey();
        }
    }
    
    class A : IObservable<int>
    {
        readonly Subject<int> observers = new Subject<int>();
    
        public IDisposable Subscribe(IObserver<int> observer)
        {
            return observers.Subscribe(observer);
        }
    
        public void Notify(int x) => observers.OnNext(x);
    }
    
[/code]

すると、出力結果は

[code]

    Main Thread Id: 1
    
[/code]

となりました。

期待していた動作としては、

[code]

    Main Thread Id: 1
    Current Thread Id: X, Value:1
    Current Thread Id: X, Value:2
    Current Thread Id: X, Value:3
    
[/code]

だったのですが、`ThreadPoolScheduler` を指定することで何も実行されなくなるのが何故かわかりません。

どなたか、ご教授いただければ幸いです。  
よろしくお願い致します。


CC BY-SA 4.0
2018-12-18T05:05:46.417
51324
2019-01-24T10:56:35.613
2018-12-18T06:19:10.603
30812
30812
question
c# reactive-programming
Reactive.Extensions の SubscribeOn の挙動がわかりません


私も最初混乱したのですが、  
`SubscribeOn` は `IObserver.OnNext` ではなく`IObservable.Subscribe` の実行スケジュール指定です。  
サンプルコードでは `A` の `Subscribe` がスレッドプールから実行されます。  
`OnNext` をスケジュール指定したい場合は `ObserveOn` です。

サンプルコードの実行順は

[code]

    a.SubscribeOn(ThreadPoolScheduler.Instance).Subscribe(/**/);
    
    // この時点でラムダ式は a.observers には登録されていない。ラムダ式は呼ばれない。
    a.Notify(1);
    a.Notify(2);
    a.Notify(3); 
    
    // ここでスレッドが切り替わったと考えられる
    Console.ReadKey();
    
    // ここで a.observers に登録される
    a.observers.Subscribe(observer);
    
    // この後 Nofity は呼ばれない。ラムダ式も実行されない
    
[/code]

であるため何も表示されなかったと考えられます。


CC BY-SA 4.0
2019-01-24T10:56:35.613
52286
2019-01-24T10:56:35.613
20885
51324
answer