Cold Observable的Scheduler.CurrentThread - 它会在哪个线程中运行？

Posted 2021-03-27

我是RX的新手并开始理解一些我在网上找不到的概念。

我在我的代码中使用了一个计时器：

Observable.Timer(TimeSpan.FromSeconds(2), schedulerProvider.CurrentThread);
// where schedulerProvider.CurrentThread is actually Scheduler.CurrentThread

据我所知，Timer是一个Cold Observable，每个订阅者都会获得一个私有计时器，它将在订阅时运行2秒钟。

我正在thread1创建那个冷的观察点并用thread2订阅它[，thread3，thread4等...]

所以问题是：它将在哪个线程中运行？

请提供消息来源。

谢谢！

编辑：谢谢你的详细解答。现在我知道我的问题有一个问题。作为第二个参数传递的调度程序不是用于订阅者调度，而是用于运行Timer。默认情况下，大多数RX运算符都在同一个线程上运行，但Timer是一个（多个）异常。默认情况下，计时器本身将在线程池上运行，并在同一线程上调用OnNext。如果要求它使用CurrentThread调度程序，则计时器本身将在同一个线程上运行（与大多数Rx运算符一样），并且将在同一线程上调用订阅者。 （对于常规用例，这不是很推荐）。

答案

为什么要运行一些测试才能找到答案？

这是我做的：

Console.WriteLine(Thread.CurrentThread.ManagedThreadId);

Observable
    .Timer(TimeSpan.FromSeconds(2.0), Scheduler.CurrentThread)
    .Subscribe(x => Console.WriteLine(Thread.CurrentThread.ManagedThreadId));

制作：

12
12

然后我尝试了这个：

Console.WriteLine(Thread.CurrentThread.ManagedThreadId);

Observable
    .Timer(TimeSpan.FromSeconds(2.0))
    .Subscribe(x => Console.WriteLine(Thread.CurrentThread.ManagedThreadId));

那产生了：

12
13

Console.WriteLine(Thread.CurrentThread.ManagedThreadId);

Observable
    .Timer(TimeSpan.FromSeconds(2.0))
    .ObserveOn(Scheduler.CurrentThread)
    .Subscribe(x => Console.WriteLine(Thread.CurrentThread.ManagedThreadId));

那产生了：

11
27

所有这些归结为Scheduler.CurrentThread在评估Scheduler.CurrentThread参数时捕获当前线程上下文。

在第一个代码块中，它是在创建计时器时捕获的 - 换句话说，就是我的控制台线程。

在最后一个块中，它在计时器触发后被捕获，因此它捕获了计时器触发的线程。