Java9 基于异步响应式流的发布-订阅框架

逃离我推掉我的手 2022-06-07 02:58 328阅读 0赞

> 为响应式流(Reactive Streams)增加的发布-订阅(publisher-subscriber)框架、并发包`CompletableFuture`类的增强，等等。。

JEP266中为Java语言的并发性又引入许多新的方式：响应式流，一个为它而生互操作性更强的发布-订阅框架；并且为了Java9其他API而增强的 `java.util.concurrent.CompletableFuture` 类， 以及其他的更多的更新。

在本文中，展开对响应式流的介绍，然后介绍这个发布订阅框架。

**响应式流(Reactive Streams)**

批处理系统在收集了足够多的数据，达到某一个阈值亟待进行下一步操作的时候，就衍生出了一个新的名词—数据处理(Data processing)。这时候，面向流(stream-oriented)的架构思想可以帮助我们尽快达成这个目标。它可以捕获和处理实时数据，并且可以快速地(秒级甚至更短)基于处理的结果来对系统进行相应的操作。和它相比，一个批处理系统可能会花费数秒、数天、甚至更久来做出响应。

处理数据流（特别是大小不定的实时数据）需要在异步系统中特别小心。主要问题是要控制资源消耗，避免数据源和处理系统出现供大于求(积压)的情况。这时候，需要异步地来对数据进行并行处理，利用分布式系统或者发挥多核CPU的效能，能有效地使数据处理过程变得快速高效。

响应式流(Reactive Streams)为这种非阻塞背压的异步流处理提供了一个标准。在处理系统出现过载的时候，采用异步发送信号的方式通知数据源做相应的处理。这个通知的信号就像是水管的阀门一样，关闭这个阀门会增加背压(数据源对处理系统的压力)，同时也会增加处理系统的压力。

这个标准的目的是治理跨异步边界的流数据交换(比如向其他线程传输数据) ，同时确保处理系统不被缓冲数据而压垮。换一种说法，背压是这个标准模型的一个组成部分，以便允许在线程之间调停的队列被界定。特别注意，背压通信是异步的。

响应式流(Reactive Streams)的提出就致力于提供一组最小规模的接口、方法、或者协议来描述这个操作或实体：具有非阻塞背压的异步数据流。

**发布-订阅(publisher-subscriber)框架**

Java 9 通过`java.util.concurrent.Flow` 和`java.util.concurrent.SubmissionPublisher` 类来实现响应式流。

`Flow` 类中定义了四个嵌套的静态接口，用于建立流量控制的组件，发布者在其中生成一个或多个供订阅者使用的数据项：

*  Publisher：数据项发布者、生产者
 *  Subscriber：数据项订阅者、消费者
 *  Subscription：发布者与订阅者之间的关系纽带，订阅令牌
 *  Processor：数据处理器

发布者(Publisher)以流的方式发布数据项，并注册订阅者，并且实现 `Flow.Publisher` 接口，该接口声明了一个方法，我们通过调用它来为发布者注册订阅者：

void subscribe(Flow.Subscriber<? super T> subscriber)

调用此方法来向发布者注册订阅者，但是，如果此订阅者已被其他发布者注册或注册失败(策略冲突)，这个方法就会调用订阅者的`onError()` 方法来抛出`IllegalStateException` 异常，除此之外，订阅者的`onSubscribe()` 方法会调用一个新的`Flow.Subscription` ，当空对象传给订阅者时，`subscribe()` 方法会抛出NullPointerException异常。

订阅者(Subscriber)从订阅的发布者中返回数据项，并且实现`Flow.Subscriber<T>` ，这个接口声明的方法如下：

void onSubscribe(Flow.Subscription subscription)
    void onComplete()
    void onError(Throwable throwable)
    void onNext(T item)

`onSubscribe()` 方法用来确认订阅者注册到发布者是否注册成功，它以参数列表的方式接收一个`Flow.Subscription`类型的参数，而这个参数类型里面声明的方法允许向发布者请求发布新的数据项，或请求发布者不再发布更多的数据项。

`onComplete()` 方法用在当订阅者没有调用其他方法，而`Subscription` 发生错误没有终止的情况下。调用这个方法之后，此订阅者就不能调用其他方法。

`onError(Throwable throwable)` 方法用在当发布者或订阅者遭遇不可恢复的错误的时候， 调用这个方法之后，此订阅者也不能调用其他方法。

`onNext()` 方法用于声明下一个数据项的订阅，如果在此过程中抛出异常，结果将得不到确认，甚至会导致订阅被取消。

一个订阅令牌(Subscription)为发布者和订阅者定义一种关系， 使得订阅者接收特定的数据项或者在特定时间取消接收请求，订阅令牌实现自`Flow.Subscription` 接口，该接口声明方法如下：

void request(long n)
    void cancel()

`request()` 方法添加n个数据项到当前未满的订阅请求中。如果n小于或等于0，订阅者的`onError()` 方法会被调用，并且抛出IllegalArgumentException 异常，此外，如果n大于0，订阅者就会在`onNext()` 方法的调用下接收到n个数据项，除非中间异常终止。 从Long.MAX\_VALUE次到n次中间是无界的调用。

`cancel()` 用来终止订阅者接收数据项，它有一种尝试机制，也就是说，在调用它之后也有可能收到数据项。

最后，数据处理器(Processor)在不改变发布者与订阅者的情况下基于流做数据处理，可以在发布者与订阅者之间放多个数据处理器，成为一个处理器链，发布者与订阅者不依赖于数据处理，它们是单独的过程。JDK9中不提供具体的数据处理器，必须由开发者来通过实现无方法声明的`Processor`接口来自行构建。

`SubmissionPublisher` 实现自`Flow.Publisher` 接口，向当前订阅者异步提交非空的数据项，直到它被关闭。每个当前订阅者以一个相同的顺序接收新提交的数据项，除非数据项丢失或者遇到异常。`SubmissionPublisher` 允许数据项在丢失或阻塞的时候扮演发布者角色。

`SubmissionPublisher` 提供了三个构造方法来获取实例。无参的构造器依赖于 `ForkJoinPool.commonPool()` 方法来提交发布者，以此实现生产者向订阅者提供数据项的异步特性。

下面的程序演示了`SubmissionPublisher` 用法和这套发布-订阅框架的其他特性：

import java.util.Arrays;
    
    import java.util.concurrent.Flow.*;
    import java.util.concurrent.SubmissionPublisher;  
    
    public class FlowDemo { 
       public static void main(String[] args)
       {
          // Create a publisher.
    
          SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
    
          // Create a subscriber and register it with the publisher.
    
          MySubscriber<String> subscriber = new MySubscriber<>();
          publisher.subscribe(subscriber);
    
          // Publish several data items and then close the publisher.
    
          System.out.println("Publishing data items...");
          String[] items = { "jan", "feb", "mar", "apr", "may", "jun",
                             "jul", "aug", "sep", "oct", "nov", "dec" };
          Arrays.asList(items).stream().forEach(i -> publisher.submit(i));
          publisher.close();
    
          try
          {
             synchronized("A")
             {
                "A".wait();
             }
          }
          catch (InterruptedException ie)
          {
          }
       }
    }
    
    class MySubscriber<T> implements Subscriber<T>
    {
       private Subscription subscription;
    
       @Override
       public void onSubscribe(Subscription subscription)
       {
          this.subscription = subscription;
          subscription.request(1);
       }
    
       @Override
       public void onNext(T item)
       {
          System.out.println("Received: " + item);
          subscription.request(1);
       }
    
       @Override
       public void onError(Throwable t)
       {
          t.printStackTrace();
          synchronized("A")
          {
             "A".notifyAll();
          }
       }
    
       @Override
       public void onComplete()
       {
          System.out.println("Done");
          synchronized("A")
          {
             "A".notifyAll();
          }
       }
    }

其中使用了`wait()`和`notifyAll()` 方法来使主线程等到`onComplete()` 的完成，否则是不会看到任何输出的。

下面是输出结果：

Publishing data items...
    Received: jan
    Received: feb
    Received: mar
    Received: apr
    Received: may
    Received: jun
    Received: jul
    Received: aug
    Received: sep
    Received: oct
    Received: nov
    Received: dec
    Done

最后说一句，熟悉RxJava的同学可以会心一笑了。

[原文][Link 1]

[Link 1]: https://www.javaworld.com/article/3198904/learn-java/java-9s-other-new-enhancements-part-6-concurrency.html