java - 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() 用于输入输出绑定(bind)工作，它是另一个执行其工作的线程，除了它们被缓存并回收用于另一个工作(如果将来有的话)来了。

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

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

2:新消费者接受单个值的功能接口(interface)(回调)。

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

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

与 Observable 的主要区别是 new Subscriber 支持背压，而两者的工作原理几乎相同，而且 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