java Reactive Streams 响应式流式编程

作者：字母哥哥

2022 年 7 月 07 日
本文字数：2469 字
阅读完需：约 8 分钟

Java 9 的 Reactive Streams 是对异步流式编程的一种实现。它基于异步发布和订阅模型，具有非阻塞“背压”数据处理的特点。

Non-blocking Back Pressure(非阻塞背压)：它是一种机制，让发布订阅模型中的订阅者避免接收大量数据（超出其处理能力），订阅者可以异步通知发布者降低或提升数据生产发布的速率。它是响应式编程实现效果的核心特点！

一、Java9 Reactive Stream API

Java 9 提供了一组定义响应式流编程的接口。所有这些接口都作为静态内部接口定义在java.util.concurrent.Flow类里面。

下面是 Java 响应式编程中的一些重要角色和概念，先简单理解一下

发布者（Publisher）是潜在的无限数量的有序数据元素的生产者。它根据收到的需求(subscription)向当前订阅者发布一定数量的数据元素。
订阅者（Subscriber）从发布者那里订阅并接收数据元素。与发布者建立订阅关系后，发布者向订阅者发送订阅令牌（subscription），订阅者可以根据自己的处理能力请求发布者发布数据元素的数量。
订阅令牌（subscription）表示订阅者与发布者之间建立的订阅关系。当建立订阅关系后，发布者将其传递给订阅者。订阅者使用订阅令牌与发布者进行交互，例如请求数据元素的数量或取消订阅。

二、Java 响应式编程四大接口

2.1.Subscriber Interface（订阅者订阅接口）

public static interface Subscriber<T> {    public void onSubscribe(Subscription subscription);    public void onNext(T item);    public void onError(Throwable throwable);    public void onComplete();}

复制代码

onSubscribe：在发布者接受订阅者的订阅动作之后，发布任何的订阅消息之前被调用。新创建的Subscription订阅令牌对象通过此方法传递给订阅者。
onNext：下一个待处理的数据项的处理函数
onError：在发布者或订阅遇到不可恢复的错误时调用
onComplete：当没有订阅者调用（包括 onNext()方法）发生时调用。

2.2.Subscription Interface （订阅令牌接口）

订阅令牌对象通过Subscriber.onSubscribe()方法传递

public static interface Subscription {    public void request(long n);    public void cancel();}

复制代码

request(long n)是无阻塞背压概念背后的关键方法。订阅者使用它来请求 n 个以上的消费项目。这样，订阅者控制了它当前能够接收多少个数据。
cancel()由订阅者主动来取消其订阅，取消后将不会在接收到任何数据消息。

2.3.Publisher Interface(发布者接口)

@FunctionalInterfacepublic static interface Publisher<T> {    public void subscribe(Subscriber<? super T> subscriber);}

复制代码

调用该方法，建立订阅者 Subscriber 与发布者 Publisher 之间的消息订阅关系。

2.4.Processor Interface（处理器接口）

处理者 Processor 可以同时充当订阅者和发布者，起到转换发布者——订阅者管道中的元素的作用。用于将发布者 T 类型的数据元素，接收并转换为类型 R 的数据并发布。

public static interface Processor<T,R> extends Subscriber<T>, Publisher<R> {}

复制代码

二、实战案例

现在我们要去实现上面的四个接口来完成响应式编程

Subscription Interface 订阅令牌接口通常不需要我们自己编程去实现，我们只需要在知道 request()方法和 cancle()方法含义即可。
Publisher Interface 发布者接口，Java 9 已经默认为我们提供了实现 SubmissionPublisher，该实现类除了实现 Publisher 接口的方法外，提供了一个方法叫做submit()来完成消息数据的发送。
Subscriber Interface 订阅者接口，通常需要我们自己去实现。因为在数据订阅接收之后，不同的业务有不同的处理逻辑。
Processor 实际上是 Publisher Interface 和 Subscriber Interface 的集合体，有需要数据类型转换及数据处理的需求才去实现这个接口

下面的例子实现的式字符串的数据消息订阅处理

实现订阅者 Subscriber Interface

import java.util.concurrent.Flow;
public class MySubscriber implements Flow.Subscriber<String> {
  private Flow.Subscription subscription;  //订阅令牌
  @Override  public void onSubscribe(Flow.Subscription subscription) {      System.out.println("订阅关系建立onSubscribe: " + subscription);      this.subscription = subscription;      subscription.request(2);  }
  @Override  public void onNext(String item) {      System.out.println("item: " + item);      // 一个消息处理完成之后，可以继续调用subscription.request(n);向发布者要求数据发送      //subscription.request(n);  }
  @Override  public void onError(Throwable throwable) {      System.out.println("onError: " + throwable);  }
  @Override  public void onComplete() {      System.out.println("onComplete");  }}

复制代码

SubmissionPublisher 消息发布者

import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.Flow;import java.util.concurrent.SubmissionPublisher;
public class SubmissionPublisherExample {  public static void main(String[] args) throws InterruptedException {      ExecutorService executor = Executors.newFixedThreadPool(1);      SubmissionPublisher<String> sb = new SubmissionPublisher<>(executor, Flow.defaultBufferSize());      sb.subscribe(new MySubscriber());   //建立订阅关系，可以有多个订阅者      sb.submit("数据 1");  //发送消息1      sb.submit("数据 2"); //发送消息2      sb.submit("数据 3"); //发送消息3
      executor.shutdown();  }}

复制代码

控制台打印输出结果

订阅关系建立onSubscribe: java.util.concurrent.SubmissionPublisher$BufferedSubscription@27e81a39item: 数据 1item: 数据 2

复制代码

请注意：即使发布者 submit 了 3 条数据，MySubscriber 也仅收到了 2 条数据进行了处理。是因为我们在MySubscriber#onSubscribe()方法中使用了subscription.request(2);。这就是“背压”的响应式编程效果，我有能力处理多少数据，就会通知消息发布者给多少数据。

发布于: 刚刚阅读数: 3

原文链接:【http://xie.infoq.cn/article/8eb6172872c84d8fbe9594a7a】。未经作者许可，禁止转载。

字母哥哥

关注

公众号：字母哥杂谈 2018.02.09 加入

百人团队技术经理。2017年度吉林省软件行业协会“最卓越程序员奖”获得者。前微软员工，微软MVP。华为云享专家。

发布

暂无评论

创作场景

java Reactive Streams 响应式流式编程

一、Java9 Reactive Stream API

二、Java 响应式编程四大接口

2.1.Subscriber Interface（订阅者订阅接口）

2.2.Subscription Interface （订阅令牌接口）

2.3.Publisher Interface(发布者接口)

2.4.Processor Interface（处理器接口）

二、实战案例

实现订阅者 Subscriber Interface

SubmissionPublisher 消息发布者

字母哥哥

评论