緩衝多個訂戶rx.js觀察到

Question

我

var subject = new rx.Subject(); 
    var stream =  rx.Observable.fromEvent(blah, 'event') 
        .filter(blah) 
        .map(blah) 
        .subscribe(subject); 

       return subject; 

然後我通過受該會以不同的方式，並以不同的速度來處理該事件幾種不同的處理程序。
所以我在每個處理器是

subject.subscribe(async function (x) { 
     const func = self[x.eventName]; 
     if (func) { 
      await eventHandlerWrapper(self.handlerName, func, x); 
     } 
     }) 

我有兩個問題， 一）如果事件進來超快速的處理程序要同步處理它們在給我的方式，正確的順序它？ b）如果不同的處理程序以不同的速度處理事件，他們將等到最慢的處理程序完成後再提供下一個事件？還是他們會按照自己的節奏緩衝和處理？

謝謝你， [R

Answer 1

首先，對象的創建可以簡化如下：

const subject = rx.Observable.fromEvent(blah, 'event') 
       .filter(blah) 
       .map(blah) 
       .share(); 

份額方法將創建從流的主題。如果您將此主題實例返回給每個訂閱者，您將獲得相同的行爲，並且看起來更好。

a) if the events come in super fast is the handler going to process 
them synchronously and in the right order given the way I have it? 

事件將按照正確的順序依次通過整個鏈。意思是，在處理下一個值之前，通過'fromEvent'進入的事件將被貫穿整個鏈，直到您訂閱它的點爲止（除非有一個異步操作符:)）。 Ben Lesh在角度連接2015解釋了這一點：https://www.youtube.com/watch?v=KOOT7BArVHQ（你可以觀察整個談話，但是它大約在17分鐘左右，他將陣列與觀測值進行比較）。

b) if the different handlers handle the event at different speeds are 
they all going to wait till the slowest handler is through before the  
next event is provided? or will they all sort of buffer and handle at 
they're own pace? 

他們會按照自己的步調處理事件。檢查下面的例子：

let interval$ = Rx.Observable.interval(1000).share(); 

interval$.concatMap((val) => { 
    console.log('called'); 
    return Rx.Observable.of(val).delay(3000) 
    }) 
    .subscribe((val) => console.log("slow ", val)); 

interval$.subscribe((val) => console.log("fast ", val)); 

在這裏我使用間隔可觀察，我轉換成一個主題。所以它會每秒發出一個事件。我有一個訂閱正在接受一個值，處理這個值（需要2秒），然後接下一個（使用concatMap）。另一個訂閱立即處理它們。如果你運行這個代碼（jsbin在這裏：https://jsbin.com/zekalab/edit?js,console），你會看到他們都按照自己的步調處理事件。

因此，他們不會等待最慢的處理程序，它會被內部緩衝。

如果最慢的處理器比發生事件的頻率慢，那麼您描述的情況可能會有潛在的危險情況。在那種情況下，你的緩衝區會不斷增長，最終你的應用程序會崩潰。這是一個被稱爲背壓的概念。你比事件處理速度快。在這種情況下，您需要在最慢的處理器上使用像'buffer'或'window'這樣的運算符來避免這種情況。