java - RxJava 调度程序始终与 sleep 在同一线程中工作

Question

我尝试在不同的线程上运行每个计算，但无论我使用什么调度程序，它总是在单线程上运行。

PublishProcessor processor = PublishProcessor.create();

    processor
        .doOnNext(i ->System.out.println(i.toString()+" emitted on "+Thread.currentThread().getId()))
        .observeOn(Schedulers.newThread()).subscribe(i -> {
            System.out.println(i.toString()+" received on "+Thread.currentThread().getId());
            Thread.currentThread().sleep(5000);
        });
    processor.onNext(2);
    processor.onNext(3);
    processor.onNext(4);
    processor.onNext(5);
    processor.onNext(6);


    while (true) {}

输出将是:

2 emitted on 1
3 emitted on 1
4 emitted on 1
5 emitted on 1
6 emitted on 1
2 received on 13
3 received on 13
4 received on 13
5 received on 13
6 received on 13

线程 13 仅在 hibernate 后处理下一个值，但在这种情况下我希望有几个单独的 hibernate 线程。 有人可以解释一下我做错了什么吗？

Answer 1

.observeOn(...) 通过将项目流更改为另一个线程来生效，但它始终是同一个线程。

如果您想为每个项目创建一个新线程，您可以这样做

processor
    .doOnNext(i ->System.out.println(i.toString()+" emitted on "+Thread.currentThread().getId()))
    .flatMap(item -> Observable.just(item)
                         .subscribeOn(Schedulers.newThread()))    // make every item change to a new thread
    .subscribe(i -> {
        System.out.println(i.toString()+" received on "+Thread.currentThread().getId());
        Thread.currentThread().sleep(5000);
    });