如何用RxJava并行处理文本行

Question

我想知道如何用RxJava同时处理文本行。现在，我拥有的是一个来自条目集的可观察项，以及一个订阅者，该订阅者在每个条目上处理onNext调用中的条目。用户订阅带有这行代码的可观察

obs.observeOn(Schedulers.io()).subscribe(sub);

但当我运行它时，它的运行速度和顺序版本一样慢，而且似乎是按顺序处理它。我将如何使其并发？

Answer 1

您的observeOn(Schedulers.io())调用意味着将在该线程上观察到所有排放。你想让他们进入他们自己的线程。

在这里，我使用flatMap为源发出的每个项目创建一个新的可观察对象。在映射函数内部，我必须defer处理工作，直到订阅，否则整个链在处理完成时被阻塞。我还必须确保通过subscribeOn在新线程上进行订阅。

Random r = new Random();
Observable.from(new String[]{"First", "Second", "Third", "Fourth", "Fifth"})
    .flatMap(new Func1<String, Observable<String>>() {
        public Observable<String> call(final String s) {
            return Observable.defer(new Func0<Observable<String>>() {
                public Observable<String> call() {
                    Thread.sleep(r.nextInt(1000));
                    return Observable.just(s);
                }
            }).subscribeOn(Schedulers.newThread());
        }
    })
    .subscribe(new Action1<String>() {
        @Override
        public void call(String s) {
            System.out.println("Observed " + s + " on thread " + Thread.currentThread().getId());
        }
    });

这给了我类似的输出（注意无序和在不同的线程上，即并行处理）：

Observed Fourth on thread 17
Observed Second on thread 15
Observed Fifth on thread 18
Observed First on thread 14
Observed Third on thread 16