Reactive(2) 响应式流与制奶厂业务
再谈响应式
在前一篇文章从Reactive编程到“好莱坞”中,谈到了响应式的一些概念,讲的有些发散。 但仅仅还是停留在概念的层面,对于实战性的东西并没有涉及。 响应式编程强调的是异步化、面向流的处理方式,这两者也并非凭空生出,而是从大量的技术实践中总结提炼出来的概念,就比如:
为什么Web后端开发的,对 Reactive 没有感觉除了前端,Reactive 概念在大数据领域的应用其实非常的广泛了。 但是对于大多数做 Web 后端开发的人来说或许普及程度并不高,以笔者自身的感受是,码了这么些年头,除了做好代码分层之外,似乎也没有见到 Reactive可以发挥重大作用的地方。 原因就在于,在Web 后端开发领域基本是依托 HTTP协议机制实现的,这是一个相当简单的 请求 -> 应答 交互模式,客户端在发送请求后,会一直等待结果返回,也就是结果的通知是由客户端主动获取而非异步通知的,因此并不是 Reactive 的风格。 但这已经是符合用户一贯的使用方式了,绝大多数情况下并不需要做什么样的变化,此时我们对响应式的感知并不深刻。 更符合Reactive 的另外一个场景是 富客户端(Rich Application),假设在需要大量复杂的前端交互的场景下,我们可以选择将一些逻辑放在前端代码中实现。 浅显的从趋势上看, Reactive 的前景还是很明朗的,这里并不是说因为现在多数流行的编程语言中都有它的影子(比如提供了Rx风格的框架)。 Java 9 支持的 Reactive StreamJava 平台直到 JDK 9 才提供了对于 Reactive 的完整支持,而在此之前的JDK版本中,也以及存在一些有关联性的API,比如:
这些关联性API 并不是完整的 Reactive,Java 9所支持的 Reactive Stream API 来自于2013年的响应式流规范(Reactive Stream Specification)。 https://www.reactive-streams.org/ 基于这个规范中主要定义了下面几个接口: Java的响应式流接口统一定义在 java.util.concurrent.Flow接口
首先,在subscribe方法调用成功后,Subscriber的 onSubscribe(Subscription s) 方法会被触发(Subscription 表示当前的订阅关系)。 发布者会产生3种不同的消息,分别对应到 Subscriber 的3个回调方法: 数据消息:对应 onNext 方法,表示发布者产生的数据。 在上面的3种通知中,错误、结束消息都表示当前的流已经到达了终点,后面不再会有消息产生。
负压的支持 负压是响应式流定义的一种重要的能力,在上述的接口中,实质上已经提供了负压的支持。 范例下面,以一个简单的代码示例来演示 Reactive Stream API 是如何使用的。 以制奶厂为例,为了提高营收,工厂推出了一个厂家直销的业务。 顾客可以直接向厂方订购一定天数的奶制品,每天则是由工厂的服务人员送货上门。 为了模拟这个场景,我们实现的代码如下:
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()),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实现),其内部会启动一个定时器,用于模拟每天给用户发放生产的牛奶。
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 实现了一个"在线送奶" 的业务流。 扩展阅读关于Future和CompletableFuture的区别 (编辑:李大同) 【声明】本站内容均来自网络,其相关言论仅代表作者个人观点,不代表本站立场。若无意侵犯到您的权利,请及时与联系站长删除相关内容! |