一篇入门reactive streams背压响应流编程

Posted 2023-02-02 aofengdaxia

概念

node.js广泛流行的一个原因，就是node.js采用了非阻塞的异步编程技术， 使得node.js可以提高服务器的性能,轻松的处理大量的并发请求。

但是在异步编程中存在很多问题，比如回调地狱，错误处理等。流式编程的模式提出，可以降低异步编程的难度。许多java流式编程的类库和框架应用而生，
react streams是一种背压响应式编程规范，它通过一组最小的接口，方法和协议，使得异步的流式编程相对规范统一。 其中jdk9中的reactive stream就是react streams的java实现。

流式编程

流式编程是一种编程范式，它将数据视为流，而不是一次性的数据集合。流式编程的核心思想是，将数据集合视为一系列的元素，而不是一次性的数据集合。
数据的处理过程，就是对数据流的处理过程。数据像水流一样，从一个环节流向另一个环节，每个环节都对数据进行处理，最终得到最终的结果。
而不是像以前的编程范式通过一个方法对其他对象方法的调用完数据的操作。
流式编程采用的底层思想是订阅者-发布模式，数据的处理环节以订阅者的形式订阅数据，数据产生后，发布者会将数据推送给订阅者。
订阅者一次异步的处理数据，而不是一次性的处理所有数据。我写的一篇关于观察者模式的文章 可以帮助更好的理解这种编程模式。

背压

背压是指数据的生产者和消费者之间的一种协调机制，当数据的生产者生产数据的速度大于消费者消费数据的速度时，就会出现背压。react streams的背压协议规定了当订阅者无法处理数据时，发布者应该暂停数据的发布。
当订阅者可以处理数据时，发布者应该继续发布数据。这样就可以保证数据的生产者和消费者之间的速度一致，避免数据的丢失。
可以理解为一种根据订阅者处理能力的自适应的数据反馈发布机制。

核心

react streams的核心是Publisher, Subscriber, Subscription, Processor。

• Publisher: 发布者，它负责发布数据，发布者可以是一个数据源，也可以是一个数据处理器。
• Subscriber: 订阅者，它负责订阅数据，订阅者可以是一个数据处理器，也可以是一个数据终端。
• Subscription: 订阅，它负责订阅发布者的数据，订阅者可以订阅多个发布者。
• Processor: 处理器，它负责处理数据，处理器可以是一个数据源，也可以是一个数据终端。

通过使用类库提供的这四个核心的库，或者定制自己的具体实现，就可以很容易的使用背压响应式编程。

publisher

public interface Publisher<T> 
    void subscribe(Subscriber<? super T> s);

Publisher是一个发布者，它负责发布数据，发布者可以是一个数据源，也可以是一个数据处理器。
其中的subscribe方法，就是订阅者订阅发布者的入口。
可以通过实现publisher接口，其中jdk9提供了Flow.Publisher接口的一些常用的实现，比如：

public class MyPublisher implements Publisher<String> 
    @Override
    public void subscribe(Subscriber<? super String> s) 
        s.onSubscribe(new MySubscription(s));
    

jdk9提供的常用的publisher实现

• SubmissionPublisher: 一个可提交的发布者，它可以通过submit方法提交数据，当数据提交后，会将数据推送给订阅者。

subscriber

public interface Subscriber<T> 
    void onSubscribe(Subscription s);
    void onNext(T t);
    void onError(Throwable t);
    void onComplete();

Subscriber是一个订阅者，它负责订阅数据，订阅者可以是一个数据处理器，也可以是一个数据终端。
其中的onSubscribe方法，是当订阅者订阅发布者后，发布者会调用订阅者的onSubscribe方法，将订阅者的订阅对象传递给订阅者。
其中的onNext方法，是当发布者发布数据后，会调用订阅者的onNext方法，将数据传递给订阅者。
其中的onError方法，是当发布者发布数据出现异常后，会调用订阅者的onError方法，将异常传递给订阅者。
其中的onComplete方法，是当发布者发布数据完成后，会调用订阅者的onComplete方法，通知订阅者数据发布完成。
一般情况，需要自行实现subscriber接口，来完成对数据的处理，比如：

public class MySubscriber implements Subscriber<String> 
    private Subscription subscription;
    @Override
    public void onSubscribe(Subscription s) 
        this.subscription = s;
        this.subscription.request(1);
    
    @Override
    public void onNext(String s) 
        System.out.println("onNext: " + s);
        this.subscription.request(1);
    
    @Override
    public void onError(Throwable t) 
        t.printStackTrace();
    
    @Override
    public void onComplete() 
        System.out.println("onComplete");
    

subscription

public interface Subscription 
    void request(long n);
    void cancel();

Subscription是一个订阅，它负责订阅发布者的数据，订阅者可以订阅多个发布者。
其中的request方法中的n参数，是订阅者请求发布者发布数据的数量。
其中的cancel方法，是方便订阅者取消订阅发布者的数据。

processor

public interface Processor<T, R> extends Subscriber<T>, Publisher<R> 

Processor是一个处理器，它负责处理数据，处理器可以是一个数据源，也可以是一个数据终端。
从接口的定义上不难看出，Processor是一个发布者和订阅者的组合，它既可以发布数据，也可以订阅数据。

public class MyProcessor implements Processor<String, String> 
    private Subscriber<? super String> subscriber;
    @Override
    public void subscribe(Subscriber<? super String> s) 
        this.subscriber = s;
        this.subscriber.onSubscribe(new MySubscription(s));
    
    @Override
    public void onSubscribe(Subscription s) 
        s.request(1);
    
    @Override
    public void onNext(String s) 
        System.out.println("onNext: " + s);
        this.subscriber.onNext(s);
    
    @Override
    public void onError(Throwable t) 
        t.printStackTrace();
    
    @Override
    public void onComplete() 
        System.out.println("onComplete");
    

当publisher产生数据后，只可以流向一个subscriber。
如果需要依次流向多个subscriber接收数据，就需要使用processor，processor可以作为中间的处理和传递的环节，因为他即可以是一个publisher，也可以是一个subscriber。

综合写法

import java.io.IOException;
import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

/**
 * @Author zhangshiyu
 * @Date 2023/1/26 18:25
 * @project reactive-stream-demo
 */
public class ReactiveStreamDemo 
    //main
    public static void main(String[] args) throws IOException 

        //创建发布者
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
        //创建订阅者
        Flow.Subscriber<String> subscriber = new Flow.Subscriber<String>() 
            @Override
            public void onSubscribe(Flow.Subscription subscription) 
                //订阅者处理订阅请求
                System.out.println("onSubscribe");
                subscription.request(Integer.MAX_VALUE);
            

            @Override
            public void onNext(String s) 
                //订阅者处理发布者发布的消息
                System.out.println("onNext----\\r\\n" + s);
            

            @Override
            public void onError(Throwable throwable) 
                //订阅者处理发布者发布的异常消息
                System.out.println("onError");
            

            @Override
            public void onComplete() 
                //订阅者处理发布者发布的完成消息
                System.out.println("onComplete");
            
        ;
        Flow.Processor<String, String> processor = new Flow.Processor<String, String>() 
            private Flow.Subscriber<? super String> subscriber;

            @Override
            public void subscribe(Flow.Subscriber<? super String> subscriber) 
                this.subscriber = subscriber;
            

            @Override
            public void onSubscribe(Flow.Subscription subscription) 
                subscriber.onSubscribe(subscription);
            

            @Override
            public void onNext(String item) 
                System.out.println("processor----" + item);
                subscriber.onNext(item + "(processed)");
            

            @Override
            public void onError(Throwable throwable) 
                subscriber.onError(throwable);
                System.out.println("processor----\\r\\n" + throwable.getMessage());
            

            @Override
            public void onComplete() 
                subscriber.onComplete();
                System.out.println("processor----\\r\\n" + "onComplete");
            
        ;
        //发布者和订阅者建立订阅关系
        processor.subscribe(subscriber);
        publisher.subscribe(processor);


        //发布者发布消息
        publisher.submit("hello");
        publisher.submit("world");
        publisher.submit("reactive");
        publisher.close();
        System.in.read();