java - RxJava subscribeOn 和observeOn 以及阻塞w/flatMap。谁做了什么？

Question

这是我第一次尝试 RxJava。

我正在尝试用于一项简单的任务:需要通过 Java 应用程序从数据库导出数据。导出分三步完成:

1. 发出查询以查找需要导出的对象的主键。
2. 批处理这些 ID，并并行获取/编码完整对象。
3. 将编码对象写入输出流。

我认为一个好的方法是让一个线程执行步骤 1(迭代 ResultSet 的页面)，让一组线程执行步骤 2(每个 ResultSet 页面一个任务)，并让主线程执行步骤 3(这需要在单个线程上发生)。

我知道在 Rx 世界中阻塞主线程并不被视为一件好事，但对于这个问题，让我们忽略这一点。这是我第一次在遗留应用程序上引入响应式(Reactive)编程。

上述场景的测试如下所示:

   @Test
   public void testSimplePipelineRx()
   {
      Scheduler idsScheduler = makeScheduler("idsExecutor");
      Scheduler dataScheduler = makeScheduler("dataExecutor");

      final List<MutablePair<Integer, List<Integer>>> stateHolder = new CopyOnWriteArrayList<>();

      Observable<Integer> idsObservable = Observable.create(SyncOnSubscribe.createSingleState(
            () ->
            {
               print("ids observable initialized");
               stateHolder.add(emitIds());
               return stateHolder.get(0);
            },
            (state, observer) ->
            {
               if (state.getLeft() >= state.getRight().size())
               {
                  print("ids observable next - emitting onComplete");
                  observer.onCompleted();
               }
               else
               {
                  Integer val = state.getRight().get(state.getLeft());
                  state.setLeft(state.getLeft() + 1);

                  print("ids observable next - emitting " + val);
                  observer.onNext(val);
               }
            },
            (state) ->
            {
               print("ids observable finish");
               state.setLeft(-1);
               state.getRight().clear();
            }
      ));

      final ConcurrentHashMap<String, Boolean> results = new ConcurrentHashMap<>();

      print("Starting");

      idsObservable
            .buffer(2)
            .flatMap(i -> Observable.just(i)
                           .observeOn(dataScheduler)
                           .map(k ->
                                {
                                   print("Transforming values: " + k.get(0) + "-" + k.get(k.size() - 1));
                                   return "Values: " + k.get(0) + "-" + k.get(k.size() - 1);
                                })
                  , 5 //max count flatMap will have queued up
            )
            .subscribeOn(idsScheduler)
            .toBlocking()
            .subscribe(new Subscriber<String>()
            {
               @Override
               public void onStart()
               {
                  request(5);
               }

               @Override
               public void onCompleted()
               {
                  print("Observed done");
               }

               @Override
               public void onError(Throwable err)
               {
                  print(ExceptionUtils.getStackTrace(err));
                  print("Observed error");
               }

               @Override
               public void onNext(String str)
               {
                  print("Observed value " + str);
                  results.put(str, true);
                  request(1);
               }
            });

      print("Asserting");
      Assert.assertEquals(Integer.valueOf(-1), stateHolder.get(0).getLeft());
      Assert.assertEquals(0, stateHolder.get(0).getRight().size());
      Assert.assertEquals(7, results.keySet().size());
      print("Finishing");
   }

   private MutablePair<Integer, List<Integer>> emitIds()
   {
      return new MutablePair<>(0, IntStream.range(0, 13).mapToObj(i -> i).collect(Collectors.toList()));
   }

   private Scheduler makeScheduler(String name)
   {
      ExecutorService executor = Executors.newFixedThreadPool(10, new ThreadFactory()
      {
         private AtomicInteger id = new AtomicInteger(0);

         @Override
         public Thread newThread(Runnable r)
         {
            return new Thread(r, name + "-" + id.getAndIncrement());
         }
      });

      return Schedulers.from(executor);
   }

   private void print(String msg)
   {
      System.out.println(new SimpleDateFormat("HH:mm:ss.SSS").format(new Date()) + " - " + Thread.currentThread().getName() + " - " + msg);
   }

我得到的输出是这样的:

16:25:36.168 - main - Starting
16:25:36.185 - idsExecutor-0 - ids observable initialized
16:25:36.194 - idsExecutor-0 - ids observable next - emitting 0
16:25:36.194 - idsExecutor-0 - ids observable next - emitting 1
16:25:36.201 - idsExecutor-0 - ids observable next - emitting 2
16:25:36.202 - dataExecutor-0 - Transforming values: 0-1
16:25:36.202 - idsExecutor-0 - ids observable next - emitting 3
16:25:36.202 - idsExecutor-0 - ids observable next - emitting 4
16:25:36.202 - dataExecutor-1 - Transforming values: 2-3
16:25:36.202 - idsExecutor-0 - ids observable next - emitting 5
16:25:36.202 - idsExecutor-0 - ids observable next - emitting 6
16:25:36.202 - idsExecutor-0 - ids observable next - emitting 7
16:25:36.203 - idsExecutor-0 - ids observable next - emitting 8
16:25:36.203 - idsExecutor-0 - ids observable next - emitting 9
16:25:36.203 - dataExecutor-3 - Transforming values: 6-7
16:25:36.203 - dataExecutor-2 - Transforming values: 4-5
16:25:36.204 - dataExecutor-4 - Transforming values: 8-9
16:25:36.206 - main - Observed value Values: 0-1
****16:25:36.206 - dataExecutor-3 - ids observable next - emitting 10
****16:25:36.206 - dataExecutor-3 - ids observable next - emitting 11
16:25:36.206 - main - Observed value Values: 2-3
****16:25:36.206 - dataExecutor-3 - ids observable next - emitting 12
16:25:36.206 - dataExecutor-5 - Transforming values: 10-11
16:25:36.206 - main - Observed value Values: 4-5
****16:25:36.206 - dataExecutor-3 - ids observable next - emitting onComplete
16:25:36.207 - main - Observed value Values: 6-7
****16:25:36.207 - dataExecutor-3 - ids observable finish
16:25:36.207 - main - Observed value Values: 8-9
16:25:36.207 - dataExecutor-6 - Transforming values: 12-12
16:25:36.207 - main - Observed value Values: 10-11
16:25:36.208 - main - Observed value Values: 12-12
16:25:36.208 - main - Observed done
16:25:36.208 - main - Asserting
16:25:36.208 - main - Finishing

为什么“dataExecutor”线程在最后接管“发射”值？我本以为“idsExecutor-0”线程是唯一一个“发出”值的线程。

Answer 1

这是 RxJava 和 observeOn 中基于协同例程的背压设计的效果。运算符:来自 observeOn 的发出线程的请求调用可以执行 SyncOnSubscribe 内的生成器(这称为弱流水线)。确保SyncOnSubscribe从已知线程生成项目，使用 subscribeOn紧接着它(这称为强管道):

Observable.range(1, 5)
.subscribeOn(Schedulers.io())
.map(v -> Thread.currentThread() + "|" + v)
.observeOn(Schedulers.single())
.subscribe(w -> Thread.currentThread() + "||" + w);

Thread.sleep(1000);