java - Observable.from(需要在另一个线程上完成的事情)

Question

我正在学习 Rx-java 并遇到一个小问题。我正在尝试获取对象列表并将其传递给 Observable.from() 以便我可以对其进行处理。

问题:这个列表需要在另一个线程上获取(http 内部调用)，所以 Observable.from(getList()) 没有给我任何东西。

我尝试过这样的事情:

Observable.create(new Observable.OnSubscribe>() { @覆盖 公共(public)无效调用(订阅者>订阅者){ 订阅者.onNext(getList()); 订阅者.onCompleted(); } }).订阅...

但是这个订阅了一个Iterable，onNext只传递完整的列表，而不是列表中的每个OIbject。我缺少什么？我怎么做 ？谢谢

更新

这是我正在尝试做的事情(以及我如何使用 toSortedList):

Observable.from(getList())
    .subscribeOn(Schedulers.newThread())
    .observeOn(AndroidSchedulers.mainThread())
    .toSortedList(new Func2<Object, Object, Integer>() {
      @Override
      public Integer call(Object left, Object right) {
        if (left.equals(right))
          return 0;
        return left.getLabel(pm).compareToIgnoreCase(right.getLabel(pm));
      }
    })
    .subscribe(new Observer<List<Object>>() {
//On next sends a List of objects I'm using
}

问题是 getList() 需要在另一个线程上调用(并且它返回一个 Iterable，而不是 Observable)。也许我没有正确使用你的第一个解决方案 dwursteisen。

更新2

这是我不明白为什么它不起作用的原因:

Observable.just(0)
    .subscribeOn(Schedulers.newThread())
    .observeOn(AndroidSchedulers.mainThread())
    .flatMap(new Func1<Integer, Observable<Object>>() {
      @Override
      public Observable<Object> call(Integer integer) {
        return Observable.from(getList());
      }
    })
    .toSortedList(new Func2<Object, Object, Integer>() {
      @Override
      public Integer call(Object left, Object right) {
        if (left.equals(right))
          return 0;
        return left.getLabel(pm).compareToIgnoreCase(right.getLabel(pm));
      }
    })
    .subscribe( ... )

或者用订阅替换 flatMap 调用并在 onNext 中执行所有操作，它仍然在主线程上调用 getList() ...

Answer 1

当您想使用特定线程时，可以使用 subscribeOn 方法来指定要使用哪个“tread”(调度程序)。

sourceObservable.subscribeOn(httpScheduler)
                .flatMap(Observable::from)
                .subscribe();

如果不起作用，您仍然可以使用您的解决方案。但只需迭代您的列表即可:

Observable.create(new Observable.OnSubscribe<CategorizedActivityInfoWrapper>() {
  @Override
  public void call(Subscriber<? super CategorizedActivityInfoWrapper> subscriber) {
     for(o : getList()) {
         subscriber.onNext(o);
     }

     subscriber.onCompleted();
  }
}).subscribe ...

更新

Observable.from(getList()).subscribe();

也可以这样写:

    List objs = getList();
    Observable.from(objs).subscribe();

因此，getList()将在当前线程中被调用。您将无法控制在哪个线程中调用 getList()