ObserveOn mit Scheduler.NewThread beobachtet nicht, wenn OnNext des Beobachters blockiert ist und fortgesetzt wird

Question

Kann jemand bitte helfen, erklären, warum, wenn ich "blockieren und fortsetzen" Beobachter onNext Sequenz einen Puffer mit Zeit beobachtbare Sequenz abonniert, dass Scheduler.NewThread nicht mehr gelten?

Zum Beispiel:

Wenn ich eine Folge von Zahlen über

var query = from number in Enumerable.Range(1,200)
            select SnoozeNumberProduction(number);

var observableQuery = query.ToObservable();
var bufferedSequence = observableQuery.Buffer(TimeSpan.FromSeconds(2));

Wobei SnoozeNumberProduction die Nummerngenerierung um 250 ms verzögert

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

Wenn ich nun später die gepufferte Sequenz mit einem "ObserveOn(Scheduler.NewThread)" so abonniere, dass ich auf dem vierten Puffer mit einem Console.ReadKey blockiere

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]);
});

In diesem Fall, wenn ich nach etwa 10 Sekunden eine Taste drücke, sehe ich, dass die folgenden Puffer auf demselben ManagedThread ausgeführt werden, selbst wenn Scheduler.NewThread in ObserveOn erwähnt wird. Kann jemand bitte dieses Verhalten erklären?

Beispielhafte Ausgabe:

(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**

Answer 1

ObserveOn ist selbst eine Schicht in Ihrer zusammengesetzten Sequenz, deren einzige Aufgabe es ist, zu einem anderen Scheduler zu wechseln. Ihre Schlafphasen finden jedoch in einer Select auftretend auf einem IEnumerable . Diese Sequenz wird dann in eine IObservable mit ToObservable die standardmäßig auf Dispatcher.CurrentThread .

Erst jetzt wechseln Sie für jeden Artikel, der hereinkommt, in einen anderen Thread. Wenn Sie es zu ändern:

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

var bufferedSequence = query.Buffer(TimeSpan.FromSeconds(2));

Jetzt ist die Aufzählung tritt in einem neuen Thread auf, und da Sie nichts tun, um das zu ändern, wird es dort bleiben.

Es gibt sogar eine Observable.Range die als IObservable und nimmt eine optionale IDispatcher . Allerdings habe ich angenommen, dass Ihre Quelle nicht wirklich ist Enumerable.Range . Falls ja, hier ist das Äquivalent:

var query = from number in Observable.Range(1,200, Dispatcher.NewThread)
            select SnoozeNumberProduction(number);

var bufferedSequence = query.Buffer(TimeSpan.FromSeconds(2));

Answer 2

Ich habe diese Frage im MSDN Rx-Forum gepostet http://social.msdn.microsoft.com/Forums/en-US/rx/thread/52e72a11-9841-4571-b86d-f805d3aeb7b5 und erfuhr, dass dies aus Gründen der Effizienz geschieht

---

Sie blockieren den Aufruf von OnNext in Subscribe. Der ObserveOn-Operator stellt sicher, dass OnNext so oft wie möglich auf dem aktuellen Thread aufgerufen wird. Der ObserveOn-Operator verwendet den aktuellen Thread erneut, um OnNext nacheinander für so viele Werte aufzurufen, wie derzeit verfügbar sind. Da Sie in Subscribe blockieren, sammeln sich mehrere Aufrufe von OnNext an. Nachdem Sie die Blockierung aufgehoben haben, werden die Aufrufe in der Warteschlange auf demselben Thread ausgeführt. Damit soll der Overhead vermieden werden, der durch die Erstellung eines neuen Threads pro Benachrichtigung entsteht, wenn dies nicht erforderlich ist.