专栏首页Java3yReactive响应式流入门!

Reactive响应式流入门!

再谈响应式

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

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

  • 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 请求的数据量,以此保证自身不会被冲垮。

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

范例

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

以某一个制奶厂为例,为了提高营收,工厂推出了一个厂家直销的业务。 顾客可以直接向厂方订购一定天数的奶制品,每天则是由工厂的服务人员送奶上门。为了模拟这个场景,我们实现的代码如下:

  • 制奶厂,一个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()方法可以将数据推送给用户。

  • 顾客,一个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入参,即表示订购的数量,在首次订阅时会请求第一天的奶品,此后则每次收到到奶品后再请求下一天的,直到将总量消费完。

  • 测试程序

执行下面的代码:

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 来构建的,后面有机会再做下介绍。

本文分享自微信公众号 - Java3y(java3y)

原文出处及转载信息见文内详细说明,如有侵权,请联系 yunjia_community@tencent.com 删除。

原始发表时间:2019-11-04

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 纳税服务系统一(用户模块)【简单增删改查、日期组件、上传和修改头像】

    前言 为了更好地掌握SSH的用法,使用一个纳税服务系统来练手…..搭建SSH框架环境在上一篇已经详细地说明了。http://blog.csdn.net/hon_...

    Java3y
  • 常用的CSS框架

    常用的CSS框架 之前在写自己的个人网站的时候,由于自己Web前端不是特别好,于是就去找相关的CSS框架来搭建页面了。 找到以下这么一篇文章(列出了很多常用的C...

    Java3y
  • 从react 编程 到 "好莱坞"

    Reactive Programming(响应式编程)已经不是一个新东西了。关于 Reactive 其实是一个泛化的概念,由于很抽象,一些理论性的介绍很容易把人...

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

    在前一篇文章 从Reactive编程到“好莱坞” 中,谈到了响应式的一些概念,讲的有些发散。但仅仅还是停留在概念的层面,对于实战性的东西并没有涉及。所以大家看了...

    美码师
  • Reactive Programming 一种技术,各自表述

    作为一名 Java 开发人员,尤其是 Java 服务端工程师,对于 Reactive Programming 的概念似乎相对陌生。随着 Java 9 以及 Sp...

    kirito-moe
  • iOS开发——Carthage安装和使用教程

    Carthage 使用于 Swift 语言编写,只支持动态框架,只支持 iOS8+的Cocoa依赖管理工具。

    Originalee
  • 为何与Face++互怼百度推广?人脸识别路在何方?云从CEO全解答

    牧北 若朴 采访/整理 量子位·QbitAI 出品 “世界上的四家人脸识别的独角兽都在中国。这个市场没有大到可以支撑四个独角兽”,李开复说这是中国特色。 从另一...

    量子位
  • Prometheus vs Zabbix

    公司要上监控,Prometheus 是最热门的监控解决方案,作为喜新厌旧的程序员,我当然是选择跟风了,但上级更倾向于 Zabbix,那没办法,只能好好对比一番,...

    用户2443625
  • html --- bootstrap 框架 (栅格系统布局)

    小蔚
  • 高效幂模算法探究:Montgomery算法解析

    模运算,又称模算数(modular arithmetic),是一个整数的算术系统,其中数字超过一定值后(称为模)会“卷回”到较小的数值,模运算最早是卡尔·弗里德...

    FB客服

扫码关注云+社区

领取腾讯云代金券