将 Observables 收集到列表中似乎不会立即发出集合

我使用 RxJava 来收集单独发出的 Observables 列表，并将它们组合成一个 Observables 列表（本质上与 flatMap 相反）。这是我的代码：

        // myEvent.findMemberships() returns an Observable<List<Membership>>

        myEvent.findMemberships()
             .flatMap(new Func1<List<Membership>, Observable<User>>() {
               @Override
               public Observable<User> call(List<Membership> memberships) {
                 List<User> users = new ArrayList<User>();
                 for (Membership membership : memberships) {
                   users.add(membership.getUser());
                 }
                 return Observable.from(users);
               }
             })
             .toList()
             .subscribeOn(Schedulers.newThread())
             .observeOn(AndroidSchedulers.mainThread())
             .subscribe(new Observer<List<User>>() {
               @Override
               public void onCompleted() { }

               @Override
               public void onError(Throwable e) {
                 Timber.e(e, "Error when trying to get memberships");
               }

               @Override
               public void onNext(List<User> users) {
                 Timber.d("%d users emitted", users.size());
               }
             })

我注意到我的 onNext 从未被调用。我似乎无法理解这一点。如果我删除 .toList 调用并基本上输出各个用户（如下所示），它会通过发出每个项目来工作。

subscriptions //
    .add(currentEvent.findMemberships()
             .flatMap(new Func1<List<Membership>, Observable<User>>() {
               @Override
               public Observable<User> call(List<Membership> memberships) {
                 List<User> users = new ArrayList<User>();
                 for (Membership membership : memberships) {
                   users.add(membership.getUser());
                 }
                 return Observable.from(users);
               }
             })
             .subscribeOn(Schedulers.newThread())
             .observeOn(AndroidSchedulers.mainThread())
             .subscribe(new Observer<User>() {
               @Override
               public void onCompleted() { }

               @Override
               public void onError(Throwable e) {
                 Timber.e(e, "Error when trying to get memberships");
               }

               @Override
               public void onNext(User user) {
                 Timber.d("%d users emitted", user.getName());
               }
             }));

Q1.我对 .toList 的理解不正确吗？

Q2。如何组合一串单独发射的流Observable<Object>变成一个单一的Observable<List<Object>> ?

** EDIT

@kjones 完全解决了这个问题。我没有通过 findMemberships 调用来调用 onComplete 。我在下面添加了代码片段。我的真实用例有点复杂，有更多的转换，这就是为什么我需要使用 .toList 调用。正如 @zsxwing 也正确指出的那样，对于这个用例，一个简单的地图就足够了。

public Observable<List<Membership>> findMemberships() {
return Observable.create(new Observable.OnSubscribe<List<Membership>>() {
  @Override
  public void call(Subscriber<? super List<Membership>> subscriber) {
    try {
      // .....
      List<Membership> memberships = queryMyDb();

      subscriber.onNext(memberships);

      // BELOW STATEMENT FIXES THE PROBLEM ><
      // subscriber.onCompleted();

    } catch (SQLException e) {
      // ...
    }
  }
});

}

看起来像由myEvent.findMemberships()call 永远不会调用 onComplete。你能显示这段代码吗？

如果是这种情况，就可以解释您所看到的行为。这.toList()在所有项目都已发出（由 onComplete 发出信号）之前，不会发出列表。

你的第二个版本没有.toList()，将按如下方式进行：

.findMemberships()
    emits a single List<Membership>
.flatMap()
    transforms List<Membership> into a single List<User>
    Observable.from(users) creates an observable that emits each user
.subscribe()
    onNext() is called for each user
    onCompleted() is never called.

您的原始版本：

.findMemberships()
    emits a single List<Membership>
.flatMap()
    transforms List<Membership> into a single List<User>
    Observable.from(users) creates an observable that emits each user
.toList()
    buffers each User waiting for onCompleted() to be called
    onCompleted is never called because the .findMemberships Observable never completes

有几种解决方案：

1) 使findMemberShips()Observable 调用 onComplete。如果 Observable 返回的是findMemberShips()是一个 Rx 主题（PublishSubject、BehaviourSubject 等）

2) Use Observable.just()代替Observable.from()。你已经有一个List<User>在 .flatMap() 中，只需返回它即可。使用Observable.from(users)创建一个发出每个用户的 Observable。Observable.just(users)会创建一个发出单个信号的 ObservableList<User>。不需要.toList().

3) Use .map()代替.flatMap()。再次强调，不需要.toList()。由于每个List<Membership>转变为List<User>你只需要使用.map().

myEvent
    .findMemberships()
    .map(new Func1<List<Membership>, List<User>>() {
        @Override
        public List<User> call(List<Membership> memberships) {
            List<User> users = new ArrayList<User>();
            for (Membership membership : memberships) {
                users.add(membership.getUser());
            }
            return users;
         }
    })