How to use the new feature of java9 Reactive Stream responsive programming API

1. Java9 Reactive Stream API

Java 9 provides a set of interfaces that define reactive stream programming. All these interfaces are defined in the java.util.concurrent.Flow class as static internal interfaces.

The following are some important roles and concepts in Java reactive programming. Let’s briefly understand it first

Publisher (Publisher) is a potentially unlimited number Producer of ordered data elements. It publishes a certain number of data elements to current subscribers based on received demand (subscription).

Subscriber subscribes to and receives data elements from the publisher. After establishing a subscription relationship with the publisher, the publisher sends a subscription token (subscription) to the subscriber, and the subscriber can request the number of data elements published by the publisher based on its own processing capabilities.

Subscription token (subscription) represents the subscription relationship established between the subscriber and the publisher. When a subscription relationship is established, the publisher passes it to the subscriber. Subscribers use the subscription token to interact with the publisher, such as requesting the number of data elements or unsubscribing.

2. The four major interfaces of Java responsive programming

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: When the publisher accepts Called after the subscriber's subscription action and before publishing any subscription messages. The newly created Subscription subscription token object is passed to the subscriber through this method.

onNext: The processing function of the next data item to be processed

onError: When the publisher or subscription encounters an unrecoverable error Call

onComplete: Called when no subscriber calls (including onNext() method) occur.

2.2.Subscription Interface (Subscription Token Interface)

The subscription token object is passed through the Subscriber.onSubscribe() method

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

request(long n) is the key method behind the concept of non-blocking backpressure. Subscribers use it to request more than n consumption items. In this way, the subscriber controls how much data it can currently receive. cancel()The subscriber takes the initiative to cancel his subscription. After cancellation, he will not receive any data messages.

2.3.Publisher Interface

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

Call this method to establish the message subscription relationship between the Subscriber and the Publisher.

2.4.Processor Interface

The Processor can act as both a subscriber and a publisher, and plays the role of converting elements in the publisher-subscriber pipeline. . Used to receive and convert data elements of type T from the publisher into data of type R and publish them.

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

2. Practical cases

Now we have to implement the above four interfaces to complete reactive programming

Subscription InterfaceSubscription token interface Usually we do not need to program ourselves to implement it, we only need to know the meaning of the request() method and cancel() method.

Publisher InterfacePublisher interface, Java 9 has provided us with the implementation of SubmissionPublisher by default. In addition to the method of implementing the Publisher interface, this implementation class provides a method called submit() to complete Sending of message data.

Subscriber InterfaceSubscriber interface usually needs to be implemented by ourselves. Because after data subscription is received, different businesses have different processing logic.

Processor is actually a collection of Publisher Interface and Subscriber Interface. This interface needs to be implemented only when data type conversion and data processing are required.

The following example is implemented String data message subscription processing

Implementation of subscriber 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 message publisher

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();
  }
}

Console printout results

Subscription relationship establishment
onSubscribe: java.util.concurrent.SubmissionPublisher$BufferedSubscription@27e81a39
item: Data 1
item: Data 2

Please note: Even if the publisher After submitting 3 pieces of data, MySubscriber only received 2 pieces of data for processing. It's because we used subscription.request(2); in the MySubscriber#onSubscribe() method. This is the reactive programming effect of "back pressure". As much data as I have the ability to process, I will notify the message publisher how much data to give.

The above is the detailed content of How to use the new feature of java9 Reactive Stream responsive programming API. For more information, please follow other related articles on the PHP Chinese website!