Spring Reactor parallel并发与线程切换实战

蔚落 2022-12-25 13:54 73阅读 0赞

[第一篇：Spring Reactor 操作符详解][Spring Reactor]  
[第二篇：Spring Reactor map与flatMap操作符 详解][Spring Reactor map_flatMap_]

## 这一篇讲Reactor 的并发处理，线程切换，背压 ##

### parallel 操作符 ###

描述：

*  并行处理操作符，和我们CPU核数有关，核数越大，并行处理的线程数越多；
 *  需要注意的是，如果我们设置的并行数大于2\*核数，其他线程不会被使用；
    
     *  从下面输出结果能看出，由于我电脑是4核8线程的，并行处理是8，即使我设置了10个线程，也不会进行线程之间切换，所以内容8和9，等上面线程处理完成之后才来处理的；

Flux.create(new Consumer<FluxSink<Integer>>() {
           @Override
           public void accept(FluxSink<Integer> fluxSink) {
               for (int i = 0; i < 10; i++) {
                   fluxSink.next(i);
               }
               fluxSink.complete();
           }
       }).parallel(10).runOn(Schedulers.parallel()).doOnNext(new Consumer<Integer>() {
           @Override
           public void accept(Integer integer) {
               System.out.println(Thread.currentThread().getName()+"==内容"+integer);
               try {
                   TimeUnit.SECONDS.sleep(2);
               } catch (InterruptedException e) {
                   e.printStackTrace();
               }
           }
       }).sequential().blockLast();
       // 输出结果
    parallel-1==内容0
    parallel-3==内容2
    parallel-2==内容1
    parallel-5==内容4
    parallel-6==内容5
    parallel-4==内容3
    parallel-7==内容6
    parallel-8==内容7
    // 隔了两秒之后输出
    parallel-1==内容8
    parallel-2==内容9

parallel并行处理设置类型：

1.  Schedulers.parallel() ：创建CPU内核数量一样多线程池；
2.  Schedulers.single()：可重用单个线程
3.  Schedulers.elastic() 无限制的弹性线程池,可以一直创建线程
4.  Schedulers.boundedElastic() 有界的弹性线程池,它会回收闲置的线程，默认是60s;它对创建的线程数做了限制，默认值为CPU内核数x 10，达到上限后，最多可提交10万个任务；
5.  Schedulers.fromExecutorService() 根据我们自定义线程池进行引用；

**需要注意**

*  在使用parallel操作符时，需要结合runOn一起使用，才能并行处理；
 *  之后parallel操作符之后的处理操作符才会，进入多线程并行处理，parallel之前的处理逻辑还是在之前的主线程中处理；
    
     *  我上面的例子中，Flux.create 在parallel操作符之前，所有发射事件操作都是在main主线程中完成；
     *  后面的doOnNext操作处在多线并行处理中，如果doOnNext之后还有操作，也是在多线程并行处理中处理；

### subscribeOn ###

描述：  
主要做线程的切换，让我们的事件处理进入指定线程池或者线程中处理；

Flux.create(new Consumer<FluxSink<Integer>>() {
          @Override
          public void accept(FluxSink<Integer> fluxSink) {
              for (int i = 0; i < 2; i++) {
                  System.out.println(Thread.currentThread().getName()+"==发射="+i);
                  fluxSink.next(i);
              }
    
              fluxSink.complete();
          }
      }).subscribeOn(Schedulers.single()).doOnNext(new Consumer<Integer>() {
          @Override
          public void accept(Integer integer) {
              System.out.println(Thread.currentThread().getName()+"==内容"+integer);
              
          }
      }).blockLast();
    // 输出内容
    single-1==发射=0
    single-1==内容0
    single-1==发射=1
    single-1==内容1

*  subscribOn ： 指定指定调度器，它会让整个事件处理流程进入切换线程处理;
 *  需要注意的是：不管怎么切换调度器，他都是单线程执行，需要多线程执行必须使用parallel+runOn

### sequential() ###

描述：  
如果在并行处理后想要恢复为“正常” 顺序流处理其余操作符链，则可以使用 sequential()；

### blockLast ###

描述：  
阻塞IO流的操作，就是等多线程执行完成之后，才会返回；  
block()，blockFirst()和blockLast()

### publishOn ###

描述：  
上游产生的事件往下游传递时，经过publishOn操作，publishOn会将下游的工作线程切换到指定调度器的线程处理；

Flux.create(new Consumer<FluxSink<Integer>>() {
          @Override
          public void accept(FluxSink<Integer> fluxSink) {
              for (int i = 0; i < 2; i++) {
                  System.out.println(Thread.currentThread().getName()+"==发射="+i);
                  fluxSink.next(i);
              }
    
              fluxSink.complete();
          }
      }).subscribeOn(Schedulers.single()).doOnNext(new Consumer<Integer>() {
          @Override
          public void accept(Integer integer) {
              System.out.println(Thread.currentThread().getName()+"==内容"+integer);
              
          }
      }).blockLast();
    // 输出内容
    main==发射=0
    single-1==内容0
    main==发射=1
    single-1==内容1

*  publishOn : 它只会切换publishOn之后的工作线程，之前的发射线程它会不切换；
 *  publishOn 在一条链路处理中可以多次使用

Flux.just("1","2")
    	  .publishOn(scheduler1)
    	  .map(i -> transform(i))
    	  .publishOn(scheduler2)
    	  .doOnNext(i -> processNext(i))
    	  .subscribe();

## 背压 ##

### onBackpressureBuffer() ###

描述：  
我还是以生产香皂流水线生产来举例子：上游流水线生产香皂，下游流水线给香皂包上盒子；上游流水线一分钟流过来一个香皂，下游接受到一个香皂之后，给这个香皂包装上盒子，这一分钟就完事了，如果你只用了10秒就完成了，你可以休息50秒；上游的生产加速了，一分钟流过程10个香皂，下游这里一分钟最多包装6个盒子，导致4个香皂堆积；如果持续这样流过来，导致下游大量堆积，下游崩溃，甚至累死；  
**那我们如何处理呢**  
背压的出现就是为了解决这个问题；

Flux.create(new Consumer<FluxSink<Integer>>() {
           @Override
           public void accept(FluxSink<Integer> fluxSink) {
               // 无限次发送
               while (true){
                   fluxSink.next(new Random().nextInt(100));
               }
           }
       }, FluxSink.OverflowStrategy.BUFFER).map(info->String.valueOf(info))
               .subscribe(new Consumer<String>() {
           @Override
           public void accept(String s) {
               // 下游处理
               try {
                   TimeUnit.MILLISECONDS.sleep(100);
               } catch (InterruptedException e) {
                   e.printStackTrace();
               }
               System.out.println(s);
           }
       });

我们在发射事件的第二个参数，就是背压策略；当上游生产事件超过下游处理事件的阈值时，我们采取的措施：

*  OverflowStrategy.IGNORE：忽略下游的处理能力，上游该怎么发还是怎么发，不采取任何措施；
 *  OverflowStrategy.ERROR ： 当下游无法跟上上游时，抛异常
 *  OverflowStrategy.DROP：丢失上游发送的事件，下游接受了6个香皂之后，后面4个香皂丢掉
 *  OverflowStrategy.LATEST：下游一直获取最新的事件，丢失老事件
 *  OverflowStrategy.BUFFER：用一个大桶先装着，下游从桶里取，如果大桶也满了，就会报OOM异常；  
    ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoYW5nZTk4NzY1NDMyMQ_size_16_color_FFFFFF_t_70][]
 *  当我们往漏斗里倒水时，漏斗的流速是均匀的；
 *  下游的处理也是均匀的，保证下游的安全；

我们接着看下这个例子：

Flux.create(new Consumer<FluxSink<Integer>>() {
                @Override
                public void accept(FluxSink<Integer> fluxSink) {
                    for (int i = 0; i < 100; i++) {
                        fluxSink.next(i);
                    }
                    fluxSink.complete();
                }
            }).subscribe(new Subscriber<Integer>() {
                @Override
                public void onSubscribe(Subscription s) {
                    s.request(10);
                }
                @Override
                public void onNext(Integer integer) {
                    System.out.println("处理："+integer);
                }
                @Override
                public void onError(Throwable t) {
                    System.out.println("错误");
                }
                @Override
                public void onComplete() {
                    System.out.println("完成");
                }
            });
            // 输出结果
    处理：0
    处理：1
    处理：2
    处理：3
    处理：4
    处理：5
    处理：6
    处理：7
    处理：8
    处理：9

*  从事件生成来看，我们发生了100事件，但是下游只处理了10个事件；
 *  其实在s.request(10);中做了处理，下游只请求了10个事件，所以上游即使发生再多的也是不会被处理的；
 *  下游可以根据自己的处理能力去上游获取事件；这是不是保护了下游，也就是起到了背压的效果了；
 *  那上游没有处理的90个事件哪里去了呢。其实reactor内部有一个集合保存了这90个事件，只是我们没有去取，如果上游生产事件太多，内部集合保存不下，也是会报OOM异常的；

**我们能不能上游生产多少，下游就进行处理，处理完成之后，上游才再生产呢，而不是上游一次性全部生产？**

我们在上游发射器FluxSink对象内部找了这样方法：

*  long requestedFromDownstream(); 返回是当前下游还有多少个事件未处理完成；
 *  requestedFromDownstream()返回0，表示下游事件处理完成了

在onSubscribe方法返回一个Subscription订阅者对象，这个对象相当于上游与下游连接的开关总闸；

*  s.cancel(); 取消与上游的连接，事件就流不到下游
 *  s.request(10); 与上游连接之后，从上游获取10个事件

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoYW5nZTk4NzY1NDMyMQ_size_16_color_FFFFFF_t_70 1][]

我们将Subscription保存出去，在下游处理完成事件之后，上游在生产10个事件，下游在去请求10个事件；

Subscription sp;
        @Test
        public void back(){
            Flux.create(new Consumer<FluxSink<Integer>>() {
                @Override
                public void accept(FluxSink<Integer> fluxSink) {
                    //初始10个事件
                    for (int i = 0; i < 10; i++) {
                        fluxSink.next(i);
                    }
                    AtomicInteger data = new AtomicInteger(0);
                    // 下游处理完成之后再生产
                    while (true){
                        if(fluxSink.requestedFromDownstream() == 0){
                            System.out.println("====================下游处理完成"+fluxSink.requestedFromDownstream());
                            for (int i = 0; i < 6; i++) {
                                System.out.println("上游====发射："+data.incrementAndGet());
                                fluxSink.next(data.get());
                            }
                            sp.request(6);
                        }
                    }
                }
            }, FluxSink.OverflowStrategy.BUFFER).map(info->String.valueOf(info))
                    .subscribe(new Subscriber<String>() {
                        @Override
                        public void onSubscribe(Subscription s) {
                            s.request(10);
                            sp = s;
                        }
                        @Override
                        public void onNext(String s) {
                            // 下游处理
                            try {
                                TimeUnit.MILLISECONDS.sleep(100);
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                            System.out.println("下游====处理："+s);
                        }
                        @Override
                        public void onError(Throwable t) {
                        }
                        @Override
                        public void onComplete() {
                            System.out.println("完成");
                        }
                    });
        }
        // 输出结果
    ====================下游处理完成0
    上游====发射：1
    上游====发射：2
    上游====发射：3
    上游====发射：4
    上游====发射：5
    上游====发射：6
    下游====处理：1
    下游====处理：2
    下游====处理：3
    下游====处理：4
    下游====处理：5
    下游====处理：6
    ====================下游处理完成0
    。。。

*  从结果来看，上游生产，下游消费，等下游处理完之后，上游才继续生产；只有就起到了背压的效果，保护了下游，同时不会造成上游不停的差生事件；

### 如果有需要，后期在进行源码的讲解 ###

[Spring Reactor]: https://blog.csdn.net/change987654321/article/details/110393508
[Spring Reactor map_flatMap_]: https://blog.csdn.net/change987654321/article/details/110438121
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoYW5nZTk4NzY1NDMyMQ_size_16_color_FFFFFF_t_70]: /images/20221120/92b90ecdda9846ff90c3aaa531874f93.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoYW5nZTk4NzY1NDMyMQ_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/2020120311110989.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoYW5nZTk4NzY1NDMyMQ==,size_16,color_FFFFFF,t_70