Reactive(2) 响应式流与制奶厂业务

目录

• 再谈响应式
  • 为何Web后端开发的，对 Reactive 没有感受
• Java 9 支持的 Reactive Stream
• 范例
• 小结
• 扩展阅读

再谈响应式

在前一篇文章从Reactive编程到“好莱坞”中，谈到了响应式的一些概念，讲的有些发散。 但仅仅仍是停留在概念的层面，对于实战性的东西并无涉及。
因此你们看了后，或许仍是有些不痛不痒。

响应式编程强调的是异步化、面向流的处理方式，这二者也并不是凭空生出，而是从大量的技术实践中总结提炼出来的概念，就好比：

• 咱们谈异步化，容易联想到 Java 异步IO(Asynchronized IO)，并且习惯于将其和 BIO、NIO等概念来作对比。 却不知，老早出现的 Swing 框架(Java UI)就已经将异步化思惟玩的很溜了，不信的能够看看其内部 Observer模式(观察者)的实现。

• 咱们谈流式处理，容易联想到 时下当红的 Flink框架。 但几乎全部的大数据分析、批处理应用都是基于流式进行处理的，好比 ETL，甚至是一个最简单的 Map Reduce 做业。

为何Web后端开发的，对 Reactive 没有感受

除了前端，Reactive 概念在大数据领域的应用其实很是的普遍了。 可是对于大多数作 Web 后端开发的人来讲或许普及程度并不高，以笔者自身的感觉是，码了这么些年头，除了作好代码分层以外，彷佛也没有见到 Reactive能够发挥重大做用的地方。 缘由就在于，在Web 后端开发领域基本是依托 HTTP协议机制实现的，这是一个至关简单的 请求 -> 应答 交互模式，客户端在发送请求后，会一直等待结果返回，也就是结果的通知是由客户端主动获取而非异步通知的，所以并非 Reactive 的风格。 但这已是符合用户一向的使用方式了，绝大多数状况下并不须要作什么样的变化，此时咱们对响应式的感知并不深入。

更符合Reactive 的另一个场景是 富客户端(Rich Application)，假设在须要大量复杂的前端交互的场景下，咱们能够选择将一些逻辑放在前端代码中实现。 此时的 Web 交互就再也不是整个页面的刷新，而是演变为客户端与服务端的"实时"双向通信，这类应用也比较广泛了，好比基于 WebSocket 实现的 聊天应用、小游戏等等。

浅显的从趋势上看， Reactive 的前景仍是很明朗的，这里并非说由于如今多数流行的编程语言中都有它的影子(好比提供了Rx风格的框架)。
而是将来的大数据处理、实时流计算会成为主流，这是环境决定的。 而这时 Reactive 这种"面向流"的编程模式无疑是很合适的。

Java 9 支持的 Reactive Stream

Java 平台直到 JDK 9 才提供了对于 Reactive 的完整支持，而在此以前的JDK版本中，也以及存在一些有关联性的API，好比：

• Future 和 CompletableFuture接口，用于实现异步计算。 后者较前者则是完善了异步结果通知、任务串行等特性。
• Stream 接口，能够将传统的集合转换为"流"的方式进行处理，好比迭代、映射转换。

这些关联性API 并非完整的 Reactive，Java 9所支持的 Reactive Stream API 来自于2013年的响应式流规范(Reactive Stream Specification)。

https://www.reactive-streams.org/

基于这个规范中主要定义了下面几个接口：

Java的响应式流接口统必定义在 java.util.concurrent.Flow接口

• Publisher
  即数据的发布者。 Publisher 接口定义了一个subscribe方法，用于添加订阅者：

• Subscriber
  指数据的订阅者。 Subscriber 接口定义了4个方法，用于针对不一样的事件做出响应。

首先，在subscribe方法调用成功后，Subscriber的 onSubscribe(Subscription s) 方法会被触发(Subscription 表示当前的订阅关系)。
此后，正常能够继续调用 Subscription 的 request(long n) 方法来向发布者请求数据，n是指最大的数据条目数。

发布者会产生3种不一样的消息，分别对应到 Subscriber 的3个回调方法：

数据消息：对应 onNext 方法，表示发布者产生的数据。
错误消息：对应 onError 方法，表示发布者产生了错误。
结束消息：对应 onComplete 方法，表示发布者已经完成了全部数据的发布。

在上面的3种通知中，错误、结束消息都表示当前的流已经到达了终点，后面再也不会有消息产生。

• Subscription
  Subscription 表示的是一个订阅关系。 能够经过该对象请求数据(request方法)，或者取消订阅(cancel方法)。

• Processor
  Processor 表示的一种特殊的对象，既是生产者，又是订阅者。

负压的支持

负压是响应式流定义的一种重要的能力，在上述的接口中，实质上已经提供了负压的支持。
Publisher 只有在收到请求以后，才会产生数据。 这就保证了 Subscriber 能够根据本身的处理能力，肯定要向 Publisher 请求的数据量，以此保证自身不会被冲垮。

范例

下面，以一个简单的代码示例来演示 Reactive Stream API 是如何使用的。

以某一个制奶厂为例，为了提升营收，工厂推出了一个厂家直销的业务。 顾客能够直接向厂方订购必定天数的奶制品，天天则是由工厂的服务人员送奶上门。
为了模拟这个场景，咱们实现的代码以下：

1. 制奶厂，一个Publisher实现：

public class MilkFactory extends SubmissionPublisher<String> {

    private final ScheduledFuture<?> periodicTask;
    private final ScheduledExecutorService scheduler;

    private static final List<String> milks = Arrays.asList("益力多", "酸牛奶", "原味奶", "低脂蛋奶", "羊奶", "甜牛奶");

    public MilkFactory() {
        super();
        //初始化定时器
        scheduler = new ScheduledThreadPoolExecutor(1);

        //每一天生产完牛奶并推送给消费者
        periodicTask = scheduler.scheduleAtFixedRate(
                () -> submit(produceMilk()), 0, 1, TimeUnit.SECONDS);
    }

    //随机生产牛奶
    private String produceMilk() {
        return milks.get((int) (Math.random() * milks.size()));
    }

    //关闭流
    public void close() {
        periodicTask.cancel(false);
        scheduler.shutdown();
        super.close();
    }
}

MilkFactory 集成自SubmissionPublisher(一个提供缓冲的Publisher实现)，其内部会启动一个定时器，用于模拟天天给用户发放生产的牛奶。
经过submit()方法能够将数据推送给用户。

1. 顾客，一个Subscriber实现：

public class MilkCustomer implements Flow.Subscriber<String> {
    private Flow.Subscription subscription;
    private AtomicInteger available = new AtomicInteger(0);
    private int dayCount;

    public MilkCustomer(int dayCount) {
         this.dayCount = dayCount;
    }
    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        this.subscription = subscription;
        //设置总量
        available.set(dayCount);

        //第一天
        subscription.request(1);
    }

    @Override
    public void onNext(String milk) {
        System.out.println("今天的牛奶到了: " + milk);

        //若是还有存量，继续请求
        if(available.decrementAndGet() > 0){
            subscription.request(1);
        }else{
            System.out.println("牛奶套餐已经派完，欢迎继续订购");
            this.subscription.cancel();
        }
    }

    @Override
    public void onError(Throwable t) {
        t.printStackTrace();
    }

    @Override
    public void onComplete() {
        System.out.println("closed.");
    }
}

MilkCustomer 接受一个dayCount入参，即表示订购的数量，在首次订阅时会请求第一天的奶品，此后则每次收到到奶品后再请求下一天的，直到将总量消费完。

1. 测试程序

执行下面的代码：

MilkFactory factory = new MilkFactory();

//订阅1周
MilkCustomer customer = new MilkCustomer(7);

factory.subscribe(customer);

输出：

今天的牛奶到了: 酸牛奶
今天的牛奶到了: 羊奶
今天的牛奶到了: 原味奶
牛奶套餐已经派完，欢迎继续订购

小结

在上例中，咱们使用 Java 提供的 Reactive Stream API 实现了一个"送奶上门" 的业务流。
整个过程相对是比较简单的，最关键的地方就在于对流式处理以及订阅关系的理解。 然而目前的 Reactive 实现尚未彻底的统一，好比 Spring WebFlux(SpringBoot 2支持) 仍然是基于 Reactor 私有API而不是 Reactive Stream API 来构建的，后面有机会再作下介绍。

扩展阅读

关于Future和CompletableFuture的区别 https://juejin.im/post/5adbf8226fb9a07aac240a67