jdk8 使用 reactive-stream 响应式流

网上有很多大佬已经写过了这个响应式流的原理，这里就不在说明了。
网上查了很多大多数都是基于jdk9实现，但公司还在用jdk8，所以就自己研究了一下，分享给大家，共同学习。

官网：http://www.reactive-streams.org
github：https://github.com/reactive-streams/reactive-streams-jvm/tree/v1.0.4

这个包之定义了响应式流接口，并在jdk9时加入到了jdk中，在jdk9中对接口进行了代码的实现，所以在jdk9之后可以直接使用（封装的真好，使用很方便），但在jdk8及之前的版本想要使用响应式流就要引入这个包并自己实现接口。

这里写了一个小demo用于练习理解响应式流

导入reactive-streams这个包
maven

<dependency>
    <groupId>org.reactivestreams</groupId>
    <artifactId>reactive-streams</artifactId>
    <version>1.0.4</version>
</dependency>

jar包提供了4个接口和1个jdk9及以上的适配器

class org.reactivestreams.FlowAdapters
interface org.reactivestreams.Processor
interface org.reactivestreams.Publisher
interface org.reactivestreams.Subscriber
interface org.reactivestreams.Subscription

1、创建用于传输的类

public class User {
    private Integer id;
    private String name;
    // 有参构造
    public User(Integer id, String name){
        this.id = id;
        this.name = name;
    }
    // 省略 get set 方法
}

2、创建订阅中心 UserSubscription

public class UserSubscription implements Subscription {

    public static final ConcurrentLinkedQueue<User> inbound = new ConcurrentLinkedQueue<>();
    private Subscriber subscriber;
    public UserSubscription(Subscriber subscriber){
        this.subscriber = subscriber;
    }

    @Override
    public void request(long l) {
        new Thread(() -> {
            User user = inbound.poll();
            subscriber.onNext(user);
        }).start();
    }

    @Override
    public void cancel() {
        System.out.println("【订阅者】取消订阅");
    }
}

3、创建发布者 UserPublisher

public class UserPublisher implements Publisher<User> {
    @Override
    public void subscribe(Subscriber<? super User> subscriber) {
        subscriber.onSubscribe(new UserSubscription(subscriber));
    }

    public void submit(User user){
        UserSubscription.inbound.offer(user);
        System.out.println("【发布者】发布消息：{id:"+user.getId()+", name:"+user.getName()+"}");
    }
}

4、创建订阅者 UserSubscriber

public class UserSubscriber implements Subscriber<User> {

    private Subscription subscription;

    /**
     * 初始化Subscription，并且表示可以开始接受订阅数据了
     * @param subscription
     */
    @Override
    public void onSubscribe(Subscription subscription) {
        this.subscription = subscription;
        System.out.println("【订阅者】订阅成功");
        // 订阅者必须发送需求信号，才能接受onNext信号
        subscription.request(1); //这里只接受一个数据元素
    }

    /**
     * 接受下一项订阅数据的回调方法
     * @param user
     */
    @Override
    public void onNext(User user) {
        boolean flag = true;
        if(user != null){
            while(flag){
                System.out.println("【订阅者】接受数据：{id:"+user.getId()+", name:"+user.getName()+"}");
                flag = false;
            }
        }
        this.subscription.request(1);
    }

    /**
     * 在Publisher或Subscriber遇到不可恢复的错误时调用此方法，Subscriber不在接受订阅消息
     *
     * 不得调用Subscription或Publisher上的任何方法
     * 目的时防止在处理完成信号期间发生Subscription或Publisher之间的周期竞争条件
     * 如果收到此信号必须考虑取消订阅
     * 目的只确保Subscriber遵从Subscription信号状态。在收到onError信号后，订阅就失效
     * @param throwable
     */
    @Override
    public void onError(Throwable throwable) {
        System.out.println("【订阅者】出现异常");
        this.subscription.cancel();
    }

    /**
     * 当接受完所有订阅数据，并且发布者已经关闭后会回调这个方法
     *
     * 不得调用Subscription或Publisher上的任何方法
     * 目的时防止在处理完成信号期间发生Subscription或Publisher之间的周期竞争条件
     * 如果收到此信号必须考虑取消订阅
     * 目的只确保Subscriber遵从Subscription信号状态。在收到onComplete信号后，订阅就失效
     */
    @Override
    public void onComplete() {
        System.out.println("【订阅者】数据接受完成");
        this.subscription.cancel();
    }

}

5、main方法

public class UserMain {
    public static void main(String[] args) throws Exception {
        // 1、创建发布者
        UserPublisher publisher = new UserPublisher();
        // 2、创建订阅者
        UserSubscriber subscriber = new UserSubscriber();
        // 3、发布者和订阅者建立订阅关系
        publisher.subscribe(subscriber);
        // 4、发布者发布消息
        for(int i = 0; i < 10; i++){
            publisher.submit(new User(i, "wpixel"));
            Thread.sleep(1000);
        }
    }
}

6、Console

【订阅者】订阅成功
【发布者】发布消息：{id:0, name:wpixel}
【订阅者】接受数据：{id:0, name:wpixel}
【订阅者】接受数据：{id:1, name:wpixel}
【发布者】发布消息：{id:1, name:wpixel}
【发布者】发布消息：{id:2, name:wpixel}
【订阅者】接受数据：{id:2, name:wpixel}
【发布者】发布消息：{id:3, name:wpixel}
【订阅者】接受数据：{id:3, name:wpixel}
【发布者】发布消息：{id:4, name:wpixel}
【订阅者】接受数据：{id:4, name:wpixel}
【发布者】发布消息：{id:5, name:wpixel}
【订阅者】接受数据：{id:5, name:wpixel}
【发布者】发布消息：{id:6, name:wpixel}
【订阅者】接受数据：{id:6, name:wpixel}
【发布者】发布消息：{id:7, name:wpixel}
【订阅者】接受数据：{id:7, name:wpixel}
【发布者】发布消息：{id:8, name:wpixel}
【订阅者】接受数据：{id:8, name:wpixel}
【发布者】发布消息：{id:9, name:wpixel}
【订阅者】接受数据：{id:9, name:wpixel}

作者是一名自由程序员，住在杭州，喜欢音乐、小说、旅行、以及编程。

P.S. 如果您喜欢这篇文章并且希望学习编程技术的话，请关注一下