使用RxJava限制吞吐量

Question

使用RxJava限制吞吐量

androidmultithreadingfunctional-programmingrx-javaobservable

6

我现在遇到的案例很难解释，因此我将写一个更简单的版本来解释这个问题。

我有一个通过Observable.from()发出文件序列的方法，这些文件由文件ArrayList定义。所有这些文件都应该上传到服务器上。为此，我有一个执行任务并返回Observable的函数。

Observable<Response> uploadFile(File file);

当我运行这段代码时，它变得很疯狂，Observable.from() 发出了所有的文件，它们一次性上传或者最多只有线程池能够处理的数量。

我想要最多同时上传2个文件。是否有任何操作符可以为我处理这个问题？

我尝试了buffer、window和其他一些操作符，但它们似乎只会一起发出两个项目，而不是保持两个并行文件上传。我还尝试在上传部分设置了一个最大线程池，但在我的情况下无法使用。

应该有一个简单的操作符来解决这个问题，对吗？我错过了什么吗？

- Ben Groot

1个回答

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- Michael · Accepted Answer

我认为所有文件都是并行上传的，因为您使用了flatMap()，它会同时执行所有转换。相反，您应该使用concatMap()，它会一个接一个地运行转换。如果要运行两个并行上传，则需要在文件可观察对象上调用window（2），然后像在代码中那样调用flatMap()。

Observable<Response> responses = 
    files
      .window(2)
      .concatMap(windowFiles ->
        windowFiles.flatMap(file -> uploadFile(file));
      );

更新:

我找到了一个更好的解决方案，完全符合您的要求。有一个重载的flatMap()函数，接受最大并发线程数。

Observable<Response> responses = 
    files
      .onBackpressureBuffer()
      .flatMap(index -> {
        return uploadFile(file).subscribeOn(Schedulers.io());
      }, 2);