multithreading - 并行执行计算量大的 map

Question

我是 ReactiveX 库的新手(我使用它的 scala 变体 RxScala)。

我有一个以高速率发出值的 Observable。我想对 Observable (map) 的所有值应用一个函数。我在 map 中使用的函数在计算上相当昂贵。

有没有办法让线程池并行计算 map 阶段？

Answer 1

是的，有一种方法可以做到这一点。

我会将流缓冲成 block 并使用 Schedulers.computation()(它使用基于大小等于线程池的 Executor)在 cpu 之间分配负载可用处理器的数量):

int chunkSize = 1000;
source
  .buffer(chunkSize)
  .flatMap(
    list -> 
      Observable
        .from(list)
        .map(expensive)
        .subscribeOn(Schedulers.computation()))
 ...

如果 map 操作的开销足够大，那么在没有 buffer 的情况下，您的性​​能可能一样好:

source
  .flatMap(
    x -> 
      Observable
        .just(x)
        .map(expensive)
        .subscribeOn(Schedulers.computation()))