android - RxJava : fromIterable is not emitting all the items

Question

我正在尝试使用 fromIterable 和 flatMap 运算符编写一个简单的代码。我只是在我的代码中将长值流映射到 Result 流。没有别的

这是我的代码。我有一个空类

    class Result {
        @Override
        public String toString() {
            return "result";
        }
    }

还有一个函数

    public Observable<Result> getResults(List<Long> requests) {
        return Observable.fromIterable(requests)
                .flatMap( aLong -> {
                    Result[] items = {new Result(), new Result()};
                    return Observable.fromIterable(Arrays.asList(items));
                });
    }       

我订阅如下

     List<Long> ids = new ArrayList<>(Arrays.asList(1L, 2L, 3L));

     getResults(ids)
    .subscribe(new DisposableObserver<Result>() {
        @Override
        public void onNext(Result item) {
            Log.d(TAG, "onNext: " + item);
        }

        @Override
        public void onComplete() {
            Log.d(TAG, "onCompleted: ");
        }

        @Override
        public void onError(Throwable e) {
            Log.e(TAG, "onError: " + e.getMessage());
        }
    });     

我的预期输出是

    LOG: onNext: result
    LOG: onNext: result
    LOG: onNext: result
    LOG: onNext: result
    LOG: onNext: result
    LOG: onNext: result
    LOG: onCompleted:         

对于 fromIterable 的每次发射，flatMap 预计会返回一个具有 2 个值的流，因此总共 6 次 onNext 然后 完成

但是我得到的是

    LOG: onNext: result
    LOG: onNext: result
    LOG: onCompleted:        

只有 2 次 onNext 被触发然后完成。剩下的 4 个值去了哪里？

但奇怪的是，我在 getResults 函数中的 fromIterable 的 doOnNext 中添加了一个 Log如下

    public Observable<Result> getResults(List<Long> requests) {
        return Observable.fromIterable(requests)
                .doOnNext(i -> Log.d(TAG, "fromIterable emitted " + i))
                .flatMap(aLong -> {
                    Result[] items = {new Result(), new Result()};
                    return Observable.fromIterable(Arrays.asList(items));
                });
    }

现在它正在发出所有值!!!!这是输出

    LOG: fromIterable emitted 1
    LOG: onNext: result
    LOG: onNext: result
    LOG: fromIterable emitted 2
    LOG: onNext: result
    LOG: onNext: result
    LOG: fromIterable emitted 3
    LOG: onNext: result
    LOG: onNext: result
    LOG: onCompleted: 

这里发生了什么？ onNext 对链进行了哪些更改以发出所有值？

Answer 1

正如@akarnokd 所解释的，您在非常短的时间内记录了相似的文本。在您的情况下，日志记录应该是唯一的，

List<Long> ids = new ArrayList<>(Arrays.asList(1L, 2L, 3L));

    getResults(ids)
            .subscribe(new DisposableObserver<Result>() {
                @Override
                public void onNext(Object item) {
                    Log.d(TAG, "onNext: called at " + System.currentTimeMillis());
                }

                @Override
                public void onComplete() {
                    Log.d(TAG, "onCompleted: ");
                }

                @Override
                public void onError(Throwable e) {
                    Log.e(TAG, "onError: " + e.getMessage());
                }
            });

或类似地在您的可观察对象中提及

  ...
  .doOnNext(i -> Log.d(TAG, "fromIterable emitted " + i))
  ...