RxJava 一个 Retrofit Network 调用订阅

Question

我正在尝试了解 RxJava 的改造。我见过很多关于订阅方法的不同例子，但找不到正确的解释。

第一个

 Observable<PostMessage> call =  service.callAPI(data);
        call.subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<PostMessage>(

                ));

第二个

Observable<PostMessage> call =   service.callAPI(data);
call.subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Consumer<PostMessage>(

        ) {
            @Override
            public void accept(PostMessage postMessage) throws Exception {

            }
        });

}

第三个

Observable<PostMessage> call =   service.callAPI(data);
call.subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new DisposableObserver<PostMessage>() {
            @Override
            public void onNext(PostMessage postMessage) {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });

}

第四个

Observable<PostMessage> call =  service.callAPI(data);
        call.subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<PostMessage>(

                ) {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(PostMessage postMessage) {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });

谁能解释一下这三种方法是什么。每个都有不同的含义还是做同样的事情？

Answer 1

1:a Schedulers.io() 用于输入输出绑定工作，它是另一个线程执行它的工作，除了它们被缓存和回收的事实如果将来有的话，再找一份工作。

1:b AndroidSchedulers.mainThread() 因为您想在主线程上接收返回的结果。

1:c new Subscriber Subscriber 订阅了 Flowable 并且是 Observer 的另一个实现。

2：新消费者 接受单个值的功能接口（回调）。

3: new DisposableObserver 也是一个观察者，但是抽象的并且允许通过实现 Disposable 来异步取消。

4: new Observer Observer 订阅了 Observable 并提供了一种接收推送通知的机制。当 Observable 完成时将调用 onCompleted() 和 onNext() 或 OnError() 并且仅调用一次。

与 Observable 的主要区别在于 new Subsciber 支持背压，而两者的工作方式几乎相同，而且 Subscriber 是 Observer 的一个实现.

订阅者和消费者的主要区别如下

Observer/Observable: The watching thread is observed by the controller. In case of an event happening, the controller is then notified and can assign the new task to a free thread from a reusable cached thread pool (or wait and cache the tasks in FIFO queue if all threads are currently busy). The worker threads implement Callable and either return successfull with the result (or a boolean value), or return with an error, in which case the controller may decide what to to (depending on the nature of error that has happended).

Producer/Consumer: The watching thread shares a BlockingQueue with the controller (event-queue) and the controller shares two with all workers (task-queue and result-queue). In case of an event, the watching thread puts a task object in the event-queue. The controller takes new tasks from the event-queue, reviews them and puts them in the task-queue. Each worker waits for new tasks and takes/consumes them from the task-queue (first come first served, managed by the queue itself), putting the results or errors back into the result-queue. Finally, the controller can retrieve the results from the result-queue and take according steps in case of errors.

来源：

https://softwareengineering.stackexchange.com/questions/286763/difference-between-consumer-producer-and-observer-observable

What is the difference between an Observer and a Subscriber?

http://reactivex.io/RxJava/javadoc/rx/schedulers/Schedulers.html

http://reactivex.io/RxJava/javadoc/io/reactivex/functions/Consumer.html

http://reactivex.io/RxJava/javadoc/io/reactivex/observers/DisposableObserver.html

RxJava 一个 Retrofit Network 调用订阅

RxJava an Retrofit Network call subscribe

java

android

retrofit2

rx-java2