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 Thread 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 以OnComplete 调用结束时，这将为Query 和Done 生成的每个值运行DoWork。我认为对于将调用订阅方法中的函数的线程没有任何保证，实际上，如果查询的所有值都是在将运行订阅的线程的同一线程上生成的。看来他们也在这样做，因此所有订阅调用都在同一个线程上进行，这很可能是为了消除许多常见的多线程错误。

所以您有两个问题，一个是您的日志记录，如果您将 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 运行，最好的选择是使用Observable.ToAsync。