RxJava 2：多个观察者未从单个 Observable 接收答案

【问题标题】：RxJava 2 : Multiple Observers not receiving from single ObservableRxJava 2：多个观察者未从单个 Observable 接收
【发布时间】：2019-01-18 23:04:41
【问题描述】：

我正在创建一个多个观察者订阅的可观察对象。但是，我只看到一个观察者 onNext 被调用。我在这里做错了什么。

        private Observable<String> getDataEmitter(String downloadFileName) {

           ObservableOnSubscribe<String> handler = emitter -> {
             String file = download(downloadFileName);
             if (file == null) {
               emitter.onError(e);
             }
             String[] tokens = file.split("\\n");
             for (int i = 0; i < tokens.length; i++) {
               emitter.onNext(token);
             }
             emitter.onComplete();
          };

          return Observable.create(handler);
        }

       public retrieve(String file) {

         final Observable<String> sourceObservable = getDataEmitter(file) 
          .flatMap(id -> {
            return Observable.from(service.find(id))
          }, Pair::of)
          .map(pair -> collect(request, pair));


         sourceObservable
           .flatMap(this::map)
           .map(this::fileFormat)
           .buffer(10)
           .subscribe(batched -> {
             System.out.println("b-1");
           }, err -> {
             System.out.println("error-1");
           },
           () -> {
             System.out.println("completed-1");
           });

        sourceObservable
          .map(pair -> format(pair))
          .subscribe(e -> {
            System.out.println("e-2" +e);
          },
          error -> System.out.println(error-2),
          () -> System.out.println("completed-2"));
        }

输出仅来自第一个订阅。也就是说，

b-1
completed-1

为什么不打印第二次订阅？我尝试将发布与连接甚至重播一起使用。但是，它没有帮助。我在这里做错了什么？

【问题讨论】：

仅出于测试目的，您是否尝试过先编写第二个 observable。它在输出中给出 e-2？

标签： java rx-java2

【解决方案1】：

您的 observable 已在 .buffer(10) 终止，因为一旦源 Observable 终止 (see the documentation)，buffers 从源 observable 收集 10 个发射并将整个发射作为批处理对象发射。你不会再观察到它的排放了。

【讨论】：