android - 在 RxJava2(Android) 中订阅与订阅？

Question

什么时候调用 subscribeWith 方法而不是简单的 subscribe ？用例是什么？

compositeDisposable.add(get()
    .observeOn(AndroidSchedulers.mainThread())
    .subscribeOn(Schedulers.io())
    .subscribe(this::handleResponse, this::handleError));

VS

   compositeDisposable.add(get()
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeOn(Schedulers.io())
              //  .subscribe(this::handleResponse, this::handleError);
                .subscribeWith(new DisposableObserver<News>() {
                    @Override public void onNext(News value) {
                        handleResponse(value);
                    }

                    @Override public void onError(Throwable e) {
                        handleError(e);
                    }

                    @Override public void onComplete() {
                       // dispose here ? why? when the whole thing will get disposed later
                       //via  compositeDisposable.dispose();  in onDestroy();
                    }
                }));

谢谢

稍后添加

根据文档，两者都返回一次性 SingleObserver 实例:

@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final <E extends SingleObserver<? super T>> E subscribeWith(E observer) {
    subscribe(observer);
    return observer;
}

@SchedulerSupport(SchedulerSupport.NONE)
public final Disposable subscribe(final Consumer<? super T> onSuccess, final Consumer<? super Throwable> onError) {
    ObjectHelper.requireNonNull(onSuccess, "onSuccess is null");
    ObjectHelper.requireNonNull(onError, "onError is null");
    ConsumerSingleObserver<T> s = new ConsumerSingleObserver<T>(onSuccess, onError);
    subscribe(s);
    return s;
}

其中 ConsumerSingleObserver 类实现了 SingleObserver 和 Disposable。

Answer 1

Observable#subscribe 解释:

在您的第一个代码 fragment 中:

.subscribe(this::handleResponse, this::handleError));

您实际上正在使用几个重载 Observable#subscribe 之一方法:

public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError)

还有一个也接受了 Action在完成时执行:

public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,
        Action onComplete) {

另一个选项允许您简单地传入 Observer (注意:void 方法) (编辑 2 - 此方法在 ObservableSource 中定义，这是 Observable 扩展的接口(interface)。)

public final void subscribe(Observer<? super T> observer)

在您问题的第二个代码 fragment 中，您使用了 subscribeWith方法只返回 Observer您传入(为了方便/缓存等):

public final <E extends Observer<? super T>> E subscribeWith(E observer)

观察者#onComplete 解释:

Observer#onComplete 在 Observable 发出流中的所有项目后被调用。
从java文档:

/**
 * Notifies the Observer that the {@link Observable} has finished sending push-based notifications.
 * <p>
 * The {@link Observable} will not call this method if it calls {@link #onError}.
 */
void onComplete();

例如，如果 get()在您的代码 fragment 中返回了 Observable发出多个 News对象，每一个都将在 Observer#onNext 中处理.您可以在此处处理每个项目。

在它们都被处理完(并假设没有发生错误)之后， onComplete会被调用。在这里您可以执行您需要做的任何额外操作(例如更新 UI)，因为您知道您已经处理了所有 News对象。

这不要与 Disposable#dispose 混淆它在可观察流结束(完成/错误)时被调用，或者由您手动终止观察(这是 CompositeDisposable 的用武之地，因为它可以帮助您一次性处理所有的 Disposable s) .

如果在您的方案中 get()将返回 Observable只发出一个项目，而不是使用 Observable ，考虑使用 io.reactivex.Single您只处理一项(在 onSuccess 中)，并且不需要指定 Action对于 onComplete :)

编辑 : 回复你的评论:

However I still do not get use of subscribeWith, you said it passes the observer for caching etc , where does it pass to? on complete? and from what I understood subscribeWith is not actually consuming the observable( or Single) right?

进一步澄清 subscribeWith解释，我的意思是它 将 消费 Observer您传递给 subscribeWith 的对象(与 subscribe 方法完全一样)但是它还会额外返回同一个 Observer 给你。在撰写本文时， subscribeWith 的实现是:

public final <E extends Observer<? super T>> E subscribeWith(E observer) {
    subscribe(observer);
    return observer;
}

因此， subscribeWith 可以 可与 subscribe 互换使用.

Can you give a use case of subscribeWith with example? I guess that will answer the question completely

subscribeWith javadoc 给出了以下用法示例:

Observable<Integer> source = Observable.range(1, 10);
CompositeDisposable composite = new CompositeDisposable();

ResourceObserver<Integer> rs = new ResourceObserver<>() {
     // ...
};

composite.add(source.subscribeWith(rs));

看这里 subscribeWith的用法将返回相同的 ResourceObserver实例化的对象。这为执行订阅和添加 ResourceObserver 提供了便利。到 CompositeDisposable在一行中(注意 ResourceObservable 实现了 Disposable 。)

编辑 2 回复第二条评论。

source.subscribeWith(rs); source.subscribe(rs); both return SingleObserver instance,

ObservableSource#subscribe(Observer <? super T> observer) 不返回 Observer .它是一个 void 方法(参见上面 Observable#subscribe 解释下的注释。)而 Observable#subscribeWith 做 返回 Observer .
如果我们使用 ObservableSource#subscribe 重写示例使用代码相反，我们必须像这样在两行中完成它:

source.subscribe(rs); //ObservableSource#subscribe is a void method so nothing will be returned
composite.add(rs);

而 Observable#subscribeWith方法使我们可以方便地在一行中完成上述操作 composite.add(source.subscribeWith(rs));
它可能会与所有看起来有些相似的重载 subscribe 方法混淆，但是它们之间存在差异(其中一些是微妙的)。查看代码和文档有助于区分它们。

编辑 3 subscribeWith 的另一个示例用例
subscribeWith当您有 Observer 的特定实现时，方法很有用。您可能想要重用。比如上面的示例代码中，就提供了 ResourceObserver的具体实现在订阅中，从而继承它的功能，同时仍然允许您处理 onNext onError 和 onComplete。

另一个示例用途:对于您问题中的示例代码，如果您想对 get() 执行相同的订阅怎么办？多处回应？

而不是复制 Consumer onNext 和 onError 跨不同类的实现，您可以做的是为例如定义一个新类。

//sample code..
public class GetNewsObserver extends DisposableObserver<News> {
    //implement your onNext, onError, onComplete.
    ....
}

现在，只要你这样做 get()请求，您只需执行以下操作即可订阅:

compositeDisposable.add(get()
    ...
    .subscribeWith(new GetNewsObserver()));

现在看代码很简单，你保持处理响应的责任分离，现在可以重用 GetNewsObserver哪里都行。