ObserveOn with Scheduler.NewThread 不会观察到这一点，如果观察者的 OnNext 被阻

本文关键字：如果 观察者 被阻 OnNext 这一点 Scheduler with NewThread 观察 ObserveOn | 更新日期: 2023-09-27 17:57:02

有人可以帮助解释为什么当我"阻止并继续"观察者的onNext序列订阅了具有时间可观察序列的缓冲区时，Scheduler.NewThread不再适用？

例如：

如果我通过以下方式缓冲数字序列

var query = from number in Enumerable.Range(1,200)
            select SnoozeNumberProduction(number);
var observableQuery = query.ToObservable();
var bufferedSequence = observableQuery.Buffer(TimeSpan.FromSeconds(2));

其中 SnoozeNumberProduction 将号码生成延迟 250 毫秒

static int SnoozeNumberProduction(Int32 number)
{
    Thread.Sleep(250);
    return number;
}

现在，如果我使用"ObserveOn（Scheduler.NewThread）"订阅bufferedSequence，这样我就可以使用Console.ReadKey阻止第四个缓冲区

Random random = new Random();
Int32 count = 0;
bufferedSequence.ObserveOn(Scheduler.NewThread).Subscribe(list =>
{
    Console.WriteLine("({0}) Numbers from {1}-{2} produced on Thread ID {3}", list.Count, list[0], list[list.Count -1], Thread.CurrentThread.ManagedThreadId);
    Thread.Sleep(1000);
    count++;
    if (count == 4)
    {
        Console.WriteLine("count reached to 4, blocking ... press any key to continue ");
        Console.ReadKey(); // Block and build up the queue
    }
    Console.WriteLine("Woken " + list[0] + " - " + list[list.Count - 1]);
});

在这种情况下，如果我在 10 秒左右后点击任何键，我会看到接下来的几个缓冲区在同一个托管线程上执行，即使 ObserveOn 中提到了 Scheduler.NewThread。有人可以帮忙解释这种行为吗？

示例输出：

(7) Numbers from 1-7 produced on Thread ID 12
Woken 1 - 7
(9) Numbers from 8-16 produced on Thread ID 14
Woken 8 - 16
(8) Numbers from 17-24 produced on Thread ID 15
Woken 17 - 24
(8) Numbers from 25-32 produced on Thread ID 16
count reached to 4, blocking ... press any key to continue
Woken 25 - 32
(8) Numbers from 33-40 produced on Thread ID **16**
Woken 33 - 40
(8) Numbers from 41-48 produced on Thread ID **16**
Woken 41 - 48
(8) Numbers from 49-56 produced on Thread ID **16**
Woken 49 - 56
(8) Numbers from 57-64 produced on Thread ID **16**
Woken 57 - 64
(8) Numbers from 65-72 produced on Thread ID **16**
Woken 65 - 72
(8) Numbers from 73-80 produced on Thread ID **16**
Woken 73 - 80
(8) Numbers from 81-88 produced on Thread ID **16**
Woken 81 - 88
(8) Numbers from 89-96 produced on Thread ID **16**

ObserveOn with Scheduler.NewThread 不会观察到这一点，如果观察者的 OnNext 被阻

ObserveOn本身就是组合序列中的一个层，唯一的工作就是交换到另一个调度程序。但是，您的睡眠发生在IEnumerable上发生的Select中。然后使用该序列转换为IObservable ToObservable ，默认为 Dispatcher.CurrentThread 。

只有在这一点上，您才会为传入的每个项目交换到另一个线程。如果将其更改为：

var query = from number in Enumerable.Range(1,200).ToObservable(Dispatcher.NewThread)
            select SnoozeNumberProduction(number);
var bufferedSequence = query.Buffer(TimeSpan.FromSeconds(2));

现在，枚举发生在新线程上，并且由于您没有执行任何更改操作，因此它将保留在那里。

实际上有一个Observable.Range从IObservable开始并采用可选的IDispatcher。但是，我假设您的消息来源实际上并不Enumerable.Range.如果是，这里是等效的：

var query = from number in Observable.Range(1,200, Dispatcher.NewThread)
            select SnoozeNumberProduction(number);
var bufferedSequence = query.Buffer(TimeSpan.FromSeconds(2));

我 http://social.msdn.microsoft.com/Forums/en-US/rx/thread/52e72a11-9841-4571-b86d-f805d3aeb7b5 将此问题交叉发布到 MSDN Rx 论坛，并了解到这是出于效率原因

您在"订阅"中阻止了对 OnNext 的呼叫。ObserveOn 运算符确保在当前线程上尽可能多次地调用 OnNext。ObserveOn 运算符重用当前线程按顺序调用 OnNext，以获取当前可用的值。 由于您在订阅中被阻止，因此对OnNext的多次调用将累积。 取消阻止后，排队的调用将在同一线程上执行。 我相信这是为了避免在不必要的情况下为每个通知创建新线程的开销。

相关文章：