NewThreadScheduler.Default将所有工作安排在同一个线程上。

Question

目前，我正试图与RX .NET进行并发，并被某些东西搞糊涂。我想并行运行四个相对较慢的任务，所以我认为NewThreadScheduler.Default是最好的选择，因为它“表示一个在单独线程上调度每个工作单元的对象”。

这是我的设置代码：

    static void Test()
    {
        Console.WriteLine("Starting. Thread {0}", Thread.CurrentThread.ManagedThreadId);

        var query = Enumerable.Range(1, 4);
        var obsQuery = query.ToObservable(NewThreadScheduler.Default);
        obsQuery.Subscribe(DoWork, Done);

        Console.WriteLine("Last line. Thread {0}", Thread.CurrentThread.ManagedThreadId);
    }

    static void DoWork(int i)
    {
        Thread.Sleep(500);
        Console.WriteLine("{0} Thread {1}", i, Thread.CurrentThread.ManagedThreadId);
    }

    static void Done()
    {
        Console.WriteLine("Done. Thread {0}", Thread.CurrentThread.ManagedThreadId);
    }

我假设每次"X线程Y“都会输出不同的线程id，但是实际输出是：

Starting. Thread 1
Last line. Thread 1
1 Thread 3
2 Thread 3
3 Thread 3
4 Thread 3
Done. Thread 3

所有的工作都是按顺序排列在同一个新线程上的，这不是我所期望的。

我想我错过了什么，但我想不出是什么。

Answer 1

一个可观察的查询有两个部分，即Query本身和Subscription。(这也是ObserveOn和SubscribeOn运算符之间的区别。)

你的Query

Enumerable
    .Range(1, 4)
    .ToObservable(NewThreadScheduler.Default);

这将创建一个可观察的值，为该系统生成默认NewThreadScheduler上的值。

您的订阅是

obsQuery.Subscribe(DoWork, Done);

这将在Query和Done以OnComplete调用结束时为Query生成的每个值运行OnComplete。我不认为订阅方法中的函数将被调用的线程有任何保证，实际上，如果查询的所有值都生成在同一个线程上，即运行订阅的线程上。看起来他们也在做，所以所有的订阅调用都是在同一个线程上进行的，这很可能是为了消除许多常见的多线程错误。

因此，您有两个问题，一个是日志记录问题，如果您将Query更改为

Enumerable
    .Range(1, 4)
    .Do(x => Console.WriteLine("Query Value {0} produced on Thread {1}", x, Thread.CurrentThread.ManagedThreadId);
    .ToObservable(NewThreadScheduler.Default);

您将看到在一个新线程上产生的每个值。

另一个问题是Rx的意图和设计。Query是一个长期运行的过程，而Subscription是一个处理结果的简短方法。如果您想以Rx的形式运行一个长时间运行的函数，那么最好的选择是使用Observable.ToAsync。