javascript - Zähle Ereignisse über einen Zeitraum und gib die Summe einmal pro Sekunde in RxJS aus

Ist es möglich, Ereignisse über einen bestimmten Zeitraum zu zählen und die Summe einmal pro Sekunde in RxJS auszugeben? Ich habe einen kontinuierlichen, niemals endenden Strom von Ereignissen. Jede Sekunde möchte ich die Gesamtzahl der Ereignisse über das letzte 5-Minuten-Fenster erhalten. Die Idee ist, dies zu verwenden, um ein Echtzeitdiagramm zu füllen.

Ich weiß, wie man das auf die traditionelle Weise macht, möchte aber wirklich verstehen, wie es mit reaktiver Programmierung gemacht wird.

Antwort #1

So würde ich es angehen.

Erstellen Sie ein Observable, das einfach die Anzahl der empfangenen Ereignisse zählt und dies als laufende Summe ausgibt (über scan). Erstellen Sie ein zweites Observable, das einfach die laufende Summe ist, verzögert um 5 Minuten. Erstellen Sie ein drittes Observable, das einfach das verzögerte Observable vom ersten Observable subtrahiert. Dies ergibt die laufende Summe der Ereignisse, die jünger als 5 Minuten sind. Erstellen Sie ein endgültiges Observable, das dieses dritte Observable einmal pro Sekunde abtastet.

const totalLast5Minutes = eventSource.publish(events => {
    const runningTotal = events
        .scan((e, total) => total + 1, 0)
        .startWith(0);
    const totalDelayed5Minutes = runningTotal
        .delay(5000 * 60)
        .startWith(0);
    return Rx.Observable
        .combineLatest(total, totalDelayed5Minutes, (t, td) => t - td);
});

// only sample the value once per second
Rx.Observable
    .interval(1000)
    .withLatestFrom(totalLast5Minutes, (interval, total) => total)
    .subscribe(total => console.log(`total=${total}`));

Antwort #2

Hier ist eine andere Möglichkeit, es zu tun, die meiner Meinung nach etwas einfacher sein könnte. Sie können windowWithTime verwenden - siehe ReactiveX- und RxJS-Dokumentation. Sie können Fenster erstellen, die sich überlappen, sodass Sie ein 5-Minuten-Fenster von Ereignissen haben können, gefolgt von einem weiteren 5-Minuten-Fenster von Ereignissen, das eine Sekunde später beginnt, und einem weiteren eine Sekunde danach usw.

Wenn Sie die Anzahl jedes dieser Fenster nehmen, erhalten Sie eine Reihe von Zählungen von 5 Minuten Ereignissen, eine Sekunde auseinander. Es würde ungefähr so aussehen:

source.windowWithTime(5 * 60000, 1000)   // create 5 minute windows, one second apart
      .flatMap(window => window.count()) // take each window and get the count once it completes (after 5 minutes)
      .subscribe(count => console.log(count));

Ich habe dies genau dafür benutzt, was Sie sagen - eine Ereignisrate über ein Zeitfenster in einem ununterbrochenen Strom von Ereignissen zu erhalten. Indem die Fenster sich überschneiden, erzeugt man einen gleitenden Durchschnitt der Aktivität im Strom.