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

再谈响应式

在前一篇文章从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

原文地址:https://blog.51cto.com/14254788/2441721

时间: 2024-10-28 10:38:18

Reactive(2) 响应式流与制奶厂业务的相关文章

Reactive Stream 响应式流

初识Reactive Stream Reactive Stream (响应式流/反应流) 是JDK9引入的一套标准,是一套基于发布/订阅模式的数据处理规范.响应式流从2013年开始,作为提供非阻塞背压的异步流处理标准的倡议. 它旨在解决处理元素流的问题--如何将元素流从发布者传递到订阅者,而不需要发布者阻塞,或订阅者有无限制的缓冲区或丢弃.更确切地说,Reactive流目的是"找到最小的一组接口,方法和协议,用来描述必要的操作和实体以实现这样的目标:以非阻塞背压方式实现数据的异步流".

JDK9新特性 Reactive Stream 响应式流

JDK9新特性 Reactive Stream 响应式流  本篇主要讲解 JDK9特性 Reactive Stream 响应式流,介绍 Reactive Stream是什么 背压是什么,以及JDK9中提供的关于Reactive Stream的接口和 2个使用案例包括如何使用Processor.  1.Reactive Stream 概念  Reactive Stream (响应式流/反应流) 是JDK9引入的一套标准,是一套基于发布/订阅模式的数据处理规范.响应式流从2013年开始,作为提供非阻

JVM平台上的响应式流(Reactive Streams)规范

// Reactive Streams // 响应式流是一个倡议,用来为具有非阻塞后压的异步流处理提供一个标准.大家努力的目标集中在运行时环境(JVM和JavaScript)和网络协议上. 注:响应式流其实就是一个规范,本文讲解的正是这个规范,且这个规范已经被引入到JDK9里了. 后压:就是下游出现了问题,得不到解决时,这个问题就会逆流而上,继而影响上游. 如果一个路口红绿灯坏了造成堵车,如果不管的话,用不了太长时间,车就会堵到上一个路口,如果再不管的话,整条路都会被赌满. // JDK9里的j

gateway&amp;reactive(响应式流)函数编程的webflux

springcloud.gateway是springcloud2的全新项目,该项目提供了一个构建在spring生态之上的API网关,包括spring5,springboot2,projectReactor.gateway旨在提高一种简单而有效的途径来转发请求,并为他们提供横切关注点,如安全性,监控/指标和弹性.在之前springcloud提供的网关是zull,zuul基于servlet2.5,使用阻塞架构,不支持长连接.zuul和negix相似,除了编程语言不同,zuul已经发布了zuul2,支

(2)响应式流——响应式Spring的道法术器

本系列文章索引:<响应式Spring的道法术器>.前情提要: 什么是响应式编程 1.2 响应式流 上一节留了一个坑--为啥不用Java Stream来进行数据流的操作? 原因在于,若将其用于响应式编程中,是有局限性的.比如如下两个需要面对的问题: Web 应用具有I/O密集的特点,I/O阻塞会带来比较大的性能损失或资源浪费,我们需要一种异步非阻塞的响应式的库,而Java Stream是一种同步API. 假设我们要搭建从数据层到前端的一个变化传递管道,可能会遇到数据层每秒上千次的数据更新,而显然

Reactive Cocoa 响应式编程开发实例讲解-中篇

上一篇文章作为开门篇讲述了Cocoa Reactive概述. 这里我们详细介绍一下CocoaReative在代码中的应用. 网上好多blog有人形容CocoaReative 中 signals是插座或者水龙头,感觉不是很好理解.我举个更贴近生活的,用电话订菜(餐馆是Signals,电话订阅是SubScriberNext). 1.概述 Create一个Signal我们视为是一个支持电话订餐的餐馆,他们有很多菜,油盐酱醋就更不用说,当一个电话打进来首先,这个Signal就开始执行,等菜做好了,菜馆要

5分钟理解 SpringBoot 响应式的核心-Reactor

目录 一.前言 二. Mono 与 Flux 构造器 三. 流计算 1. 缓冲 2. 过滤/提取 3. 转换 4. 合并 5. 合流 6. 累积 四.异常处理 五.线程调度 小结 参考阅读 一.前言 关于 响应式 Reactive,前面的两篇文章谈了不少概念,基本都离不开下面两点: 响应式编程是面向流的.异步化的开发方式 响应式是非常通用的概念,无论在前端领域.还是实时流.离线处理场景中都是适用的. 有兴趣的朋友可以看看这两篇文章: Reactive(1) 从响应式编程到"好莱坞" R

SpringBoot2.0不容错过的新特性 WebFlux响应式编程

第1章 课程介绍 课程介绍及导学 第2章 函数式编程和lambda表达式 本章介绍函数式编程的概念,和lambda表达式的基础语法,并分析了惰性求值的应用和实现.最后同意反编译字节码,重点剖析了lambda表达式的底层实现原理 第3章 Stream流编程 本章介绍jdk8里面stream流编程的重要知识点,并剖析流的运行机制和实现原理 第4章 reactive stream 响应式流 本章介绍jdk9的响应式流的开发过程,重点讲解响应式流的4个接口,以及背压的概念和jdk实现背压的关键. 第5章

Java响应式编程 Springboot WebFlux基础与实战

第1章 课程介绍课程介绍及导学1-1 导学 第2章 函数式编程和lambda表达式本章介绍函数式编程的概念,和lambda表达式的基础语法,并分析了惰性求值的应用和实现.最后同意反编译字节码,重点剖析了lambda表达式的底层实现原理2-1 概念2-2 为什么要使用函数式编程-12-3 为什么要使用函数式编程-22-4 lambda初接触-12-5 lambda初接触-22-6 jdk8接口新特性-12-7 jdk8接口新特性-22-8 jdk8接口新特性-32-9 函数接口-12-10 函数接