使用多线程RxJava进行反应式拉取

2019-10-28 02:04:13 阅读：187 来源： 互联网

标签：java-8 multithreading rx-java reactive-programming java

我正在尝试在RxJava中构建反应式拉式观察器.

我的观察者是这样的：

Observable<Command> myObs = Observable.create(s -> {
   Command command;
   int i = 0;
   do {
      command = NetworkOperation1.call(i);
      logger.info("Init command " + i);
      s.onNext(command);
      i++;
   } while (!command.isLast() && i < MAX);
   s.onCompleted();
});

我想以4个并发批处理(缓冲区)进行处理,如下所示：

myObs
    .buffer(10)
    .flatMap(batch -> {
          return Observable
                   .from(batch)
                   .subscribeOn(Schedulers.io())
                   .map(c -> {
                       Intermediate m = NetworkOperation2.call(c));
                       logger.info("Done intermediate " + m.id);
                       return m;
                   }
          }, 4);

然后,我需要以不同的大小批处理结果,如下所示：

    .buffer(25)
    .subscribeOn(Schedulers.newThread())
    .subscribe(list ->
         logger.info("Finished batch with " + list.size());

问题在于Observable中的命令是一次全部处理的,而我希望它们根据需要进行处理.

这是发生的情况的日志：(请注意,同时运行所有1000条命令,而不是根据需要调用)

Init command 0
Init command 1
Init command 2
...
Init command 999
Done intermediate 0
Done intermediate 1
...
Done intermediate 24
Finished batch with 25
Done intermediate 25
Done intermediate 26
...
Done intermediate 49
Finished batch with 25
...

问题：是否有一种方法可以暂停观察者的线程,以便它不会一次发出所有命令或类似的东西？我试过了request()运算符,但无法正常工作.

谢谢.

解决方法:

您需要背压感知源和运算符.您正在使用的运算符支持背压,但您的源不支持.

改为这样做：

myObs = Observable.range(1,1000)
    .map(i -> NetworkOperation1.call(i));

Observable.range支持背压,因此仅在需要时才会发出.

标签：java-8,multithreading,rx-java,reactive-programming,java
来源： https://codeday.me/bug/20191028/1948949.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9

使用多线程RxJava进行反应式拉取