0

当 concatMapEager 使用 1 并发时应该publisher.observerOn(scheduler).concatMap(mapper, 1)并且具有相同的行为？publisher.observerOn(scheduler).concatMapEager(mapper, 1, 1)我测试concatMap它可能使用不同的线程而不是调度程序observeOn来调用映射器。

例如

Flowable.just(1, 2, 3, 4, 5)
        .observeOn(Schedulers.io())
        .concatMapEager(i -> {
            System.out.println("1>>>>trigger in thread:" + Thread.currentThread());
            return Flowable.create(emitter -> {
                executorService.submit(() -> {
                    try {
                        Thread.sleep(3000L);
                    } catch (InterruptedException ignore) {
                    }
                    emitter.onNext("" + i);
                    emitter.onComplete();
                });
            }, BackpressureStrategy.DROP);
        }, 1, 1)
        .map(i -> "ok:" + i)
        .blockingSubscribe();

这打印

1>>>>trigger in thread:Thread[RxCachedThreadScheduler-1,5,main]
1>>>>trigger in thread:Thread[RxCachedThreadScheduler-1,5,main]
1>>>>trigger in thread:Thread[RxCachedThreadScheduler-1,5,main]
1>>>>trigger in thread:Thread[RxCachedThreadScheduler-1,5,main]
1>>>>trigger in thread:Thread[RxCachedThreadScheduler-1,5,main]

但如果将其更改为concatMap(mapper, 1)，则会打印

1>>>>trigger in thread:Thread[RxCachedThreadScheduler-1,5,main]
1>>>>trigger in thread:Thread[pool-1-thread-1,5,main]
1>>>>trigger in thread:Thread[pool-1-thread-2,5,main]
1>>>>trigger in thread:Thread[pool-1-thread-3,5,main]
1>>>>trigger in thread:Thread[pool-1-thread-4,5,main]

还应该concatMap用RxCachedThreadScheduler吗？或者是故意的。

4

0 回答 0