c# - 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

可观察查询有两部分，查询 本身和订阅。 (这也是 ObserveOn 和 SubscribeOn 运算符的区别。)

您的查询是

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

这会在该系统的默认 NewThreadScheduler 上创建一个 产生值 的可观察对象。

您的订阅是

obsQuery.Subscribe(DoWork, Done);

当 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，你最好的选择是使用 Observable.ToAsync .