响应式流中加入了Thread.sleep()发生的异常

Question

老师, 您好:
在响应式流的filter操作中如果加入了Thread.sleep() 这种让线程TIME_WATING的操作, 订阅的过程就不能继续执行, 这是为什么呢?
代码如下:

@Test
public void subscribeOnExecuter() throws InterruptedException {
   AtomicInteger k = new AtomicInteger();
   ExecutorService executorService = Executors.newFixedThreadPool(3, (runnable) -> {
       Thread thread = new Thread(runnable);
       thread.setName("subscribeOn-demo-elastic" + k);
       k.getAndIncrement();
       return thread;
   });

   Scheduler s = Schedulers.fromExecutor(executorService);

   Flux<Integer> flux = Flux.range(1, 4)
           .filter(i -> {
               log.info("filter in thread {}", Thread.currentThread().getName());
               try {
                   Thread.sleep(1000);
               } catch (InterruptedException e) {
                   e.printStackTrace();
               }
               return i % 2 == 0;
           })
           .subscribeOn(s)
           .map(i -> {
               log.info("map in thread {}", Thread.currentThread().getName());
               return i + 2;
           });
   Thread t = new Thread(() -> {
       log.info("start current thread");
       flux.log().subscribe(i -> log.info(String.valueOf(i)));
       flux.log().subscribe(i -> log.info(String.valueOf(i)));
       log.info("end current thread");
   });
   t.start();
   t.join();
}

Answer 1

Zachry_Zhang 2022-02-08 22:48:43

嗨同学你好，这个代码的问题不在于在filter中使用了Thread.sleep()，而在于Thread t执行完两次订阅操作和日志打印后，直接退出并结束了主线程，导致你的Scheduler 线程也一并退出。你可以尝试去除掉Thread.sleep()，并且增加Flux.range的上限为10000，看看这个单测是否能够跑到10000才结束。

• 提问者 weixin_慕UI5143857 #1

  去除掉Thread.sleep()，并且增加Flux.range的上限为10000，确实是可以跑完. 
  尝试了一下 如果在这个测试用例主线程最后增加一个长时间休眠等待(足够长),  也可以跑出结果
  这里可以理解为, 类似守护线程一样, 因为主线程结束了, 导致我的Scheduler 线程也一并退出是么.

• Zachry_Zhang #2

  那可能你的电脑比较快，我这边是跑不到10000的哈哈。严格意义上来讲它不是一个守护进程，这是JUnit 框架的一个特性：如果测试线程结束了那么所有子线程也会强制退出。如果你在main方法中去跑，是会等待所有子线程结束后退出的

• 提问者 weixin_慕UI5143857 回复 Zachry_Zhang #3

  好的, 理解了, 感谢老师