编程语言
首页 > 编程语言> > Java9第四篇-Reactive Stream API响应式编程

Java9第四篇-Reactive Stream API响应式编程

作者:互联网

file

我计划在后续的一段时间内,写一系列关于java 9的文章,虽然java 9 不像Java 8或者Java 11那样的核心java版本,但是还是有很多的特性值得关注。期待您能关注我,我将把java 9 写成一系列的文章,大概十篇左右。

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

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

一、Java9 Reactive Stream API

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

file

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

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

2.1.Subscriber Interface(订阅者订阅接口)

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

2.2.Subscription Interface (订阅令牌接口)

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

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

2.3.Publisher Interface(发布者接口)

@FunctionalInterface
public 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> {
}

二、实战案例

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

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

实现订阅者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 one rror(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@27e81a39
item: 数据 1
item: 数据 2

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

欢迎关注我的博客,里面有很多精品合集

觉得对您有帮助的话,帮我点赞、分享!您的支持是我不竭的创作动力! 。另外,笔者最近一段时间输出了如下的精品内容,期待您的关注。

标签:订阅,java,Stream,void,Reactive,API,发布者,public,subscription
来源: https://www.cnblogs.com/zimug/p/13829758.html