使用反應流的無源拉式背壓

Question

這是我對該主題的理解。

有出版商和訂閱者。

爲發佈者和訂閱的僞碼是一樣的東西，

Publisher{ 
    Subscriber s; 
    subscribe(Subscriber s){ 
     this.s = s; 
     s.onSubscribe(new Subscription(){ 
      onRequest(int n){ 
       List<Message> messages = service.getMessages(n); 
       s.onNext(messages); 
      } 

      onCancel(){ 
       s.onComplete(); 
      } 
     }); 
    } 
} 

Subscriber{ 
    Subscription s; 
    onSubscribe(Subscription s){ 
     this.s = s; 
     s.request(5); 
    } 

    onNext(List<Message> messages){ 
     messages.stream().parallel().map(this::process).collect(toList()); 
     s.request(5); 
    } 

    onComplete(){} 

    onError(e){} 

    private boolean process(Message m){ 
     //process message and return true/false according to whether it passed/failed. 
    } 
} 

我理解像，根據​​該應用程序的能力，則訂戶將呼叫請求。當應用程序健康時，用戶可以更快地處理並請求消息更多時間。 如果應用程序處於負載狀態，訂戶只有在處理完當前批次後纔會請求下一批次。如果處理需要時間，則更少的請求更多的消息。 消息的流程將根據應用程序的功能。

我的理解是否正確？

它與簡單循環有什麼不同？

while(true){ 
    List<Message> messages = service.getMessages(5); 
    messages.stream().parallel().map(this:process).collect(toList()); 
} 

在這種情況下，也消息的下一批次只同時處理消息後讀取。在這種情況下，當應用程序運行良好時，將讀取更多消息。如果緩慢的消息將不太頻繁地讀取。

這兩種方法有何不同？所有可用的不同類型的調度程序的差異是什麼？我不明白，這裏的優勢究竟是什麼。

更新1

好吧，我明白了活性拉動爲基礎的方法比簡單循環的一些優點。

如果訂閱者請求n項，發佈者是否需要在訂閱者上調用onNext（）？或者，如果發佈者使用n個元素的列表調用一次訂閱者（如前面的代碼片段所示），也可以嗎？如果需要進行n onNext（）調用，用戶正變得越來越複雜。

Publisher{ 
    Subscriber s; 
    subscribe(Subscriber s){ 
     this.s = s; 
     s.onSubscribe(new Subscription(){ 
      request(int n){ 
       service.getMessagesAsyc(n, (List<Message> messages) -> messages.stream().forEach(s::onNext)); 
      } 

      onCancel(){ 
       s.onComplete(); 
      } 
     }); 
    } 
} 

Subscriber{ 
    Subscription s; 
    COUNT = 5; 
    volatile int i = COUNT; 

    onSubscribe(Subscription s){ 
     this.s = s; 
     s.request(COUNT); 
    } 

    onNext(Message message){ 
     CompletableFuture.runAsync(() -> process(message)); 
     requestMessagesIfNeeded(); 
    } 

    private synchronized requestmessagesIfNeeded(){ 
     if(0 == i--){ 
      i = COUNT; 
      s.request(COUNT); 
     } 
    } 

    private boolean process(Message m){ 
     //process message and return true/false according to whether it passed/failed. 
    } 
} 

如果用戶可以通過一個n個消息列表，那麼也沒有其他優點。假設訂閱者只需要確認已成功處理的消息，使用批量確認API的第一種方法很容易實現。

Map<Boolean, List<Message>> partitioned = messages.stream().parallel().collect(partitioningBy(this::process)); 
service.ackowledge(partitioned.get(true)); 
s.request(5); 

第二種方法，我在onNext（）上分別獲得一條消息，但實現它看起來要困難得多。

Answer 1

onRequest(int n){ 
    List<Message> messages = service.getMessages(n); 
    s.onNext(messages); 
} 

這是對活性流不正確的視圖。 request告訴Publisher它可以做nonNext()調用。通常，這意味着代表源和消費者之間的活動連接的Subscription實現將處理request調用。

它與簡單循環有什麼不同？

反應流允許非阻塞消耗;你的示例會阻塞一個線程，直到getMessages()可以檢索到郵件的List。與Publisher一起使用的好處在於，無論消息源Publisher是阻塞的還是非阻塞的，您都不必以不同的方式使用事件。它爲這兩種情況提供了統一的編程模型。如果有一天getMessages()收到非阻塞變體，則下游消耗其包裝Publisher不必更改。

所有關於可用的不同類型的調度程序的差異都是？

A Scheduler表示對異步邊界的抽象：生成事件的線程和正在使用這些事件的線程。不同的Scheduler實現以不同的方式管理其線程，具體取決於這些線程的使用情況。某些線程將執行計算密集型任務，某些線程將因與非反應性源的交互而阻塞。 Schedulers類給出了各種標準實現的用途。