Synchronität in RxJS

Ich würde erwarten, dass der folgende code asynchron ausführen:

var range = Rx.Observable.range(0, 3000000);

range.subscribe(
  function(x) {},
  function(err) {},
  function() {
    console.log('Completed');
});

console.log('Hello World');

Aber das ist nicht der Fall. Es dauert eine Weile, um durch zu gehen die große Reihe von zahlen und nur, wenn es fertig ist die Ausführung wieder aufgenommen wird, können Sie versuchen, den code hier.

Ich bin verwirrt, wenn zu erwarten RxJS zu Verhalten, synchron oder asynchron. Kommt es auf die verwendete Methode? Meine bisherige Idee war, dass einmal wir sind in physikalischen Grössen/Observer-land, alles wird asynchron ausgeführt, ähnlich wie bei der Arbeit verspricht.

  • Events und callbacks sind nicht notwendigerweise asynchron. In der Tat, die meisten von DOM-Ereignissen synchron sind, außer für das laden und entladen. Wenn Sie das Feuer ein DOM event-synthetisch, ist der Listener synchron aufgerufen
InformationsquelleAutor Sergi Mansilla | 2014-01-23
  1. 21

    RxJs folgt den gleichen Regeln wie Rx.Net. Standardmäßig wird jede observable operator verwendet, der minimale Betrag der Asynchronität benötigt, um seine Arbeit zu tun. In diesem Fall Range ausführen können durch die Nummern synchron und so ist es nicht (die Dokumentation sagt Ihnen, es wird die Rx.Scheduler.currentThread standardmäßig.

    Wenn Sie wollen, um die Einführung von mehr Asynchronität als notwendig für einen Betrieb, müssen Sie sagen, eine andere Scheduler.

    Um das Verhalten, das Sie erwartet hatten, das Sie verwenden möchten Rx.Scheduler.timeout. Dieser wird, im wesentlichen, weil es zu planen, in jeder iteration über setTimeout. (In Wirklichkeit ist dies nicht einfach, den scheduler verwenden die Schnellste Methode zur Verfügung, die im browser zu planen aufgeschobene arbeiten).

    var range = Rx.Observable.range(0, 3000000, Rx.Scheduler.timeout);

    aktualisiert jsFiddle

    Beachten Sie, dass die Iteration über 3 Millionen zahlen über setTimeout nehmen fast ewig. Also vielleicht sollten wir verarbeiten Sie in Chargen von 1.000. So, hier nutzen wir das Standard-Verhalten von Range zu laufen synchron, dann batch die Werte und verwenden Sie observeOn zum ausführen des batches über unsere scheduler-timeout:

    var range = Rx.Observable
    .range(0, 3000000)
    .bufferWithCount(1000)
    .observeOn(Rx.Scheduler.timeout) //run each buffer via setTimeout
    .select(function (buffer, i) {
       console.log("processing buffer", i);
       return Rx.Observable.fromArray(buffer);
     })
    .concatAll(); //concat the buffers together

    jsFiddle Beachten Sie, dass es eine Verzögerung am Anfang, während range Kurbeln durch alle 3,000,000 Werte und bufferWithCount produziert die 3.000-arrays. Dieses Zeug ist ungewöhnlich für echte Produktions-code, in dem Sie Ihre Datenquelle ist nicht so trivial, wie Observable.range.

    FYI Versprechungen sind nicht anders in dieser Hinsicht. Wenn Sie anrufen then auf ein Versprechen, das bereits abgeschlossen ist, dass then - Funktion kann synchron laufen. Alle Versprechungen und Observablen wirklich vorhanden ist eine Schnittstelle können Sie durch die Bereitstellung von Rückrufen an, die garantiert ausgeführt, wenn die Bedingung erfüllt ist, ob die Bedingung bereits erfüllt ist oder erfüllt wird später. RxJs bietet dann viele, viele Mechanismen zu zwingen, etwas zu werden, werden asynchron ausgeführt, wenn Sie wirklich wollen es auf diese Weise. Und Methoden einzuführen, mit bestimmten timings.

    • Dies ist eine hervorragende Antwort, vielen Dank. Ich bin vertraut mit dem Verhalten Verspricht, aber ich brauche definitiv spielen mit Scheduler ein bisschen mehr.
    • Ein Hinweis auf verspricht: JQuery verspricht laufen synchron, wenn die latenten bereits resoled. Die meisten anderen Versprechen Implementierungen (z.B. q.js) immer asynchron ausgeführt.
    • Brandon - A+ kompatible Versprechen MUSS asynchron laufen soll. Gute Antwort, ansonsten..
    InformationsquelleAutor Brandon
Schreibe einen Kommentar