使用并发订阅者可观察到的RXJava2读取器

Question

我正在尝试实现一个RXJava2可观察

public class Launcher {
    public static void main(String[] args) throws Exception {

        Process process = Runtime.getRuntime().exec("ls /tmp");
        BufferedReader reader = new BufferedReader (new InputStreamReader(process.getInputStream()));

        Observable source = Observable.create(emitter -> {
            String line = "";
            while (line != null) {
                line = reader.readLine();
                if (line == null)
                    break;
                System.out.println("Engine " + Thread.currentThread().getName() + " - " + line); //process line
                emitter.onNext(line);
            }
            emitter.onComplete();
        }).subscribeOn(Schedulers.newThread());

        source.subscribe(line -> System.out.println("UI 1   " + Thread.currentThread().getName() + " - " + line));
        source.subscribe(line -> System.out.println("UI 2   " + Thread.currentThread().getName() + " - " + line));

        TimeUnit.SECONDS.sleep(10);

    }
}

onSubscribe使订阅者以并行方式收到通知。如果我没弄错的话，这意味着create（）中的lambda将为每个消费者并行执行。

因此，如果我有两个订阅者，他们每个人都获得了阅读器的一半行。Thread-1调用readLine（）得到一行Thread-2将不会得到，只有下一个。

这一切都有道理，但我一定错过了什么，因为我不知道如何：

1. 用一个线程读这些行

我查看了主题，试图将可观察的链接起来，但仍然无法理解。

编辑：我将示例更新为完全可运行的类。据我所知，这个问题是热与冷的对比。好像医生说可以观察到。创建（…） 应该创建一个冷代码，而我的代码显然表现为热代码。

后续问题：如果我添加类型参数，使其<代码>可见

Observable<String> source = Observable.create(emitter -> {...});
Observable<String> source2 = source.subscribeOn(Schedulers.newThread())




Answer 1

使用发布：

ConnectableObservable<String> o = Observable.create(emitter -> {
    try (BufferedReader reader = ...) {
        while (!emitter.isDisposed()) {
            String line = reader.readLine();
            if (line == null || line.equals("end")) {
                emitter.onComplete();
                return;          
            }
            emitter.onNext(line);
        }
    }
}).subscribeOn(Schedulers.io())
  .publish();

o.subscribe(/* consumer 1 */);
o.subscribe(/* consumer 2 */);

o.connect();