android - 如何使 Rx 请求更快地划分为单独的线程？

Question

对于每个数据对象，我正在获取更多详细信息，因此需要很长时间才能将所有数据存储在数据库中，我的问题是如何更快地组织 map(data->apiIcaSeResource.fetchDataDetail) 就像在单独的线程中划分

    apiIcaSeResource
    .fetchData("")
    .map(Datas::datas)
    .toFlowable()
    .flatMapIterable(datas->datas)
    .map(data - >apiIcaSeResource.fetchDataDetail)
    .map(dataDetail->Pair.create(dataDetail,data))
    .toList()
    .map(store::insert())
.subscribeOn(Schedulers.io())

Answer 1

您需要做的是使用 flatMap 并为每个 Observable 应用一个由固定大小的 Executor 线程池支持的调度程序。更准确地说，您必须使用 computation() 调度程序。

apiIcaSeResource
    .fetchData("")
    .map(Datas::datas)
    .flatMapIterable(datas->datas)
    .flatMap(data -> apiIcaSeResource.fetchDataDetail()
                     .subscribeOn(Schedulers.computation())
                     .map(dataDetail -> Pair.create(dataDetail,data)))

    .toList()
    .map(store::insert())

如果您不这样做，并且您使用例如 Schedulers.io()，您将同时创建大量线程，并且您将遇到严重的性能问题。特别是对于您的情况，您需要执行大量请求。

Schedulers.computation() 的问题在于，正如文档所述，

Schedulers.computation() is recommended to perform blocking, IO-bound work on this scheduler.

所以用它来进行网络调用似乎不是个好主意。

因此，另一种选择是使用 Schedulers.io() 或 Schedulers.newThread() 指定 Observable 的 Activity 线程限制>。例如，假设您希望最多有 4 个正在运行的网络调用，您可以这样做:

final static int MAX_ACTIVE_THREADS = 4;

apiIcaSeResource
    .fetchData("")
    .map(Datas::datas)
    .flatMapIterable(datas->datas)
    .flatMap(data -> apiIcaSeResource.fetchDataDetail()
                     .map(dataDetail -> Pair.create(dataDetail,data))
                     .subscribeOn(Schedulers.io()), MAX_ACTIVE_THREADS)
    .toList()
    .map(store::insert())

您可能可以做的另一个性能改进是对从第二个服务检索的数据进行批处理，例如使用 buffer() 运算符。这样，您将一次插入多行。请注意，我不希望这成为流量的瓶颈，因为网络运行肯定会变慢，但在我看来它仍然是一个很好的改进。