jdk9 flow api（reactive stream）

publisher、subscription、processor
它与函数式编程是两回事，各是各的概念。反应式编程是基于发布订阅，存在回压（背压），来感知客户端、服务端的压力（订阅者与发布者的互动）。
flow 是jdk9 对 reactive stream 的官方实现，spring5中webflux是另一种实现。
jdk9 中flow api：
image.png

demo，测试发布订阅

  1. package com.demo;
  3. import java.util.concurrent.Flow.Subscriber;
  4. import java.util.concurrent.Flow.Subscription;
  5. import java.util.concurrent.SubmissionPublisher;
  6. import java.util.concurrent.TimeUnit;
  8. /**
  9.  * 测试jdk9 flow （reactive
  10.  */
  11. public class TestJdkFlow {
  13.     public static void main(String[] args) throws InterruptedException {
  14.         // 1. 定义发布者
  15.         // 直接使用jdk自带的 submissionPublisher，它实现了 publisher 接口
  16.         SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
  18.         // 2. 定义发布者
  19.         Subscriber<String> subscriber = new Subscriber<>() {
  20.             private Subscription subscription;
  22.             @Override
  23.             public void onSubscribe(Subscription subscription) {
  24.                 System.out.println(">>>>>>>>>>> onSubscribe");
  25.                 // 保存订阅关系，需要使用它来给发布者响应
  26.                 this.subscription = subscription;
  27.                 // 请求一个数据
  28.                 subscription.request(1);
  29.             }
  31.             @Override
  32.             public void onNext(String item) {
  33.                 // 接收到一个数据，处理
  34.                 System.out.println(">>>>>>>>>>> onNext:" + item);
  35.                 // 处理完事，告诉发布者我还要请求数据
  36.                 subscription.request(1);
  37.                 // 如果已经拿到需要的数据，可以调用cancel告诉publisher发布者，自己不再需要数据了
  38.                 // subscription.cancel();
  40.             }
  42.             @Override
  43.             public void onError(Throwable throwable) {
  44.                 // 出现异常，处理
  45.                 System.out.println(">>>>>>>>>>> onError");
  46.                 // 可以告诉发布者，自己不需要接收收据了
  47.                 subscription.cancel();
  48.             }
  50.             @Override
  51.             public void onComplete() {
  52.                 // 全部数据处理完成了（发布者关闭了）
  53.                 System.out.println(">>>>>>>>>>> onComplete");
  54.             }
  55.         };
  57.         // 3. 发布者与订阅者建立订阅关系
  58.         publisher.subscribe(subscriber);
  60.         // 4. 生产数据，并发布数据
  61.         // 模拟数据
  62.         String str = "小辉又犯傻了。。。。";
  63.         publisher.submit(str);
  64.         publisher.submit(str+"...");
  66.         // 5. 结束后，关闭发布者
  67.         publisher.close();
  69. //        Thread.currentThread().join(100000);
  71.         // 线程保留，等待执行完成
  72.         TimeUnit.SECONDS.sleep(10);
  74.     }
  76. }

demo，测试发布者、处理器、订阅者式的发布订阅

  1. package com.demo;
  3. import java.util.Objects;
  4. import java.util.concurrent.Flow.Processor;
  5. import java.util.concurrent.Flow.Subscriber;
  6. import java.util.concurrent.Flow.Subscription;
  7. import java.util.concurrent.SubmissionPublisher;
  8. import java.util.concurrent.TimeUnit;
  10. /**
  11.  * 使用处理器，测试反应式编程：
  12.  * 相当于 发布者->数据处理器->订阅者
  13.  */
  14. public class TestJDKFlowProcessorDemo {
  16.     public static void main(String[] args) throws InterruptedException {
  17.         // 1. 发布者
  18.         SubmissionPublisher<Integer> publisher = new SubmissionPublisher<>();
  19.         // 2. 定义处理器，对数据进行过滤，并转换为 String 类型
  20.         TestJDKFlowProcessor processor = new TestJDKFlowProcessor();
  21.         // 3. 发布者和处理器建立订阅关系
  22.         publisher.subscribe(processor);
  23.         // 4. 定义最终的订阅者，消费String数据
  24.         Subscriber<String> subscriber = new Subscriber<>() {
  25.             private Subscription subscription;
  27.             @Override
  28.             public void onSubscribe(Subscription subscription) {
  29.                 System.out.println(">>>>>>>>>>>subscriber onSubscribe");
  30.                 // 保存订阅关系，需要使用它来给发布者响应
  31.                 this.subscription = subscription;
  32.                 // 请求一个数据
  33.                 subscription.request(1);
  34.             }
  36.             @Override
  37.             public void onNext(String item) {
  38.                 // 接收到一个数据，处理
  39.                 System.out.println(">>>>>>>>>>>subscriber onNext:" + item);
  40.                 // 处理完事，告诉发布者我还要请求数据
  41.                 subscription.request(1);
  42.                 // 如果已经拿到需要的数据，可以调用cancel告诉publisher发布者，自己不再需要数据了
  43.                 // subscription.cancel();
  45.             }
  47.             @Override
  48.             public void onError(Throwable throwable) {
  49.                 // 出现异常，处理
  50.                 System.out.println(">>>>>>>>>>>subscriber onError");
  51.                 // 可以告诉发布者，自己不需要接收收据了
  52.                 subscription.cancel();
  53.             }
  55.             @Override
  56.             public void onComplete() {
  57.                 // 全部数据处理完成了（发布者关闭了）
  58.                 System.out.println(">>>>>>>>>>>subscriber onComplete");
  59.             }
  60.         };
  61.         // 5. 处理器和最终订阅者建立订阅关系
  62.         processor.subscribe(subscriber);
  63.         // 6. 生产数据，并发布
  64.         publisher.submit(30);
  65.         publisher.submit(32);
  66.         publisher.submit(35);
  67.         publisher.submit(null); // java.lang.NullPointerException，不可发布空对象数据
  68.         // 7. 关闭发布者
  69.         publisher.close();
  70.         TimeUnit.SECONDS.sleep(10);
  71.     }
  73. }
  74. /**
  75.  * Processor 需要继承SubmissionPublisher，并实现 Processor
  76.  * 数据处理（相当于一个数据中转站）
  77.  *
  78.  * 输入源数据 Integer，过滤掉小于0的，然后转换为String发布出去
  79.  */
  80. class TestJDKFlowProcessor extends SubmissionPublisher<String> implements Processor<Integer, String> {
  82.     private Subscription subscription;
  84.     @Override
  85.     public void onSubscribe(Subscription subscription) {
  86.         System.out.println(">>>>>>>>>>>TestJDKFlowProcessor onSubscribe");
  87.         // 保存订阅关系，需要使用它来给发布者响应
  88.         this.subscription = subscription;
  89.         // 向发布者请求一个资源
  90.         subscription.request(1);
  92.     }
  94.     @Override
  95.     public void onNext(Integer item) {
  96.         System.out.println(">>>>>>>>>>>>>>>>>>TestJDKFlowProcessor onNext");
  97.         // 收到一个数据
  98.         System.out.println(">>>>>>>> 我收到了：" + item);
  99.         // 处理数据
  100.         if (Objects.isNull(item)) {
  101.             this.submit("你都不告诉我，你的年纪，是年纪一大把了吗");
  102.         } else if (item < 18) {
  103.             this.submit("小女子芳年二八，待字闺中人未识哦");
  104.         } else {
  105.             this.submit("英雄宝刀未老，老娘风韵犹存；来来来，喝完这杯，还有三杯");
  106.         }
  107.         // 还有数据要拿取，就再请求
  108.         subscription.request(1);
  109.         // 没有数据了，就可以不请求了
  110.         // subscription.cancel();
  112.     }
  114.     @Override
  115.     public void onError(Throwable throwable) {
  116.         // 出异常了
  117.         System.out.println(">>>>>>>>>>>TestJDKFlowProcessor onError");
  118.     }
  120.     @Override
  121.     public void onComplete() {
  122.         // 发布者关闭了
  123.         System.out.println(">>>>>>>>>>>TestJDKFlowProcessor onComplete");
  124.         // 关闭发布者
  125.         this.close();
  127.     }
  128. }

处理器，实现了接口 Publisher 和 Subscriber，所以既可以发布，又可以订阅。代码如下：
image.png

小结

发布者的 submit 方法是一个阻塞方法，当缓存满了，就不会产生新的数据，当订阅者消费了再产生。这样就实现了保证一定水平运行。