使用反应流的无源拉式背压

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类给出了各种标准实现的用途。