2

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

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

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

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);
}

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

Main Thread Id: 1
Current Thread Id: 1, Value:1
Current Thread Id: 1, Value:2
Current Thread Id: 1, Value:3

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

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);
}

すると、出力結果は

Main Thread Id: 1

となりました。

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

Main Thread Id: 1
Current Thread Id: X, Value:1
Current Thread Id: X, Value:2
Current Thread Id: X, Value:3

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

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

1
  • @v..snow ありがとうございます。 Commented 2019年2月18日 1:08

1 件の回答 1

1

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

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

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 は呼ばれない。ラムダ式も実行されない

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

この質問に回答するには、ログインする必要があります。

求めていた回答ではありませんか? のタグが付いた他の質問を参照する。