RXJava Observable 仅在以匿名类型订阅 Observer 时发出项目

Question

当我创建一个新的观察者作为匿名类型时它工作正常：

Observable<List<Post>> postsListObservable = mApplicationAPI.getPosts();

        postsListObservable.
                subscribeOn(Schedulers.io()).
                observeOn(AndroidSchedulers.mainThread()).subscribe( new Observer<List<Post>>() {

            @Override
            public void onSubscribe(Disposable d) {
                Log.i("ZOKa", "onSubscribe: ");
            }

            @Override
            public void onNext(List<Post> posts) {

                Log.i("ZOKa", "onNext: " + posts.size());
            }

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

            @Override
            public void onComplete() {
                Log.i("ZOKa", "onComplete: ");

            }
        });

当我将 Observer 创建为动态类型时，它不会发出数据

Observable<List<Post>> postsListObservable = mApplicationAPI.getPosts();

    postsListObservable.
            subscribeOn(Schedulers.io()).
            observeOn(AndroidSchedulers.mainThread());

    Observer<List<Post>>  observer = new Observer<List<Post>>() {

        @Override
        public void onSubscribe(Disposable d) {
            Log.i("ZOKa", "onSubscribe: ");
        }

        @Override
        public void onNext(List<Post> posts) {

            Log.i("ZOKa", "onNext: " + posts.size());
        }

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

        @Override
        public void onComplete() {
            Log.i("ZOKa", "onComplete: ");

        }
    };
    postsListObservable.subscribe(observer);

Logcat 第一个代码片段：

com.tripleService.basesetupfordi/I/ZOKa: onSubscribe: com.tripleService.basesetupfordi/I/ZOKa: onNext: 100: com.tripleService.basesetupfordi/I/ZOKa: onComplete:

Logcat 第二个：

com.tripleService.basesetupfordi/I/ZOKa: onError: null

那么，两者之间有什么区别？

Answer 1

那是因为 Operators return 新的 observable，但它们不会修改调用它们的 observable。第二个例子中的subscribeOn和observeOn对postsListObservable和observer没有影响。

以下应该有效：

Observable<List<Post>> postsListObservable = mApplicationAPI.getPosts();


    Observable<List<Post>> postsListObservable2 = postsListObservable.
            subscribeOn(Schedulers.io()).
            observeOn(AndroidSchedulers.mainThread());

    Observer<List<Post>>  observer = new Observer<List<Post>>() {
        ...
    };

    postsListObservable2.subscribe(observer);

或

Observable<List<Post>> postsListObservable = mApplicationAPI.getPosts();

    Observer<List<Post>>  observer = new Observer<List<Post>>() {
        ...
    };

    postsListObservable.
            subscribeOn(Schedulers.io()).
            observeOn(AndroidSchedulers.mainThread()).subscribe(observer);

RXJava Observable 仅在以匿名类型订阅 Observer 时发出项目

RXJava Observable emit items when subscribe with Observer as anonymous type ONLY

rx-android

rx-java2