RxAndroid：创建简单的 Hot Observable

Question

我正在创建一个 Observable，它在订阅时会发出整数。我现在的实现是这样设置的，所以订阅它的行为会从一开始就触发生成，如下所示：

private Observable createObservable() {
    return Observable.create (
        new Observable.OnSubscribe<Integer>() {

            @Override
            public void call(Subscriber<? super Integer> sub) {

                for (int i = 1; i < MAX_PROGRESS + 1; i++) {
                    sub.onNext(i);
                    SystemClock.sleep(1000);
                }
                sub.onCompleted();
            }
        }
    );
}

我的理解是这是一个冷的 Observable。我希望无论任何订阅者如何都可以生成序列，并且当订阅者确实订阅时，希望他们接收在订阅时恰好是当前的值。 IOW，把它变成一个热门的 Observable。我宁愿不继承 Observable，因为这会将它绑定到一个具体的 Integer，而实际上实际类型会有所不同。

Answer 1

在您的示例中，您每次调用函数时都使用“Observable.create”。 “热”订阅从 Observable 中保存一些实例。你还需要使用一些 Rx 方法 (cache(), retry()) 在代码中它看起来像：

    public Observable<Bitmap> mObservable;
    Subscriber<Bitmap> mSubscriber;
    Subscription mSubscription;
    Bitmap loadedBitmap;

....

@Override
    public View onCreateView(LayoutInflater inflater, ViewGroup container,
                             Bundle savedInstanceState) {

        if (savedInstanceState != null) {
            loadedBitmap = savedInstanceState.getParcelable("LoadedBitmap");
            imageView.setImageBitmap(loadedBitmap);
        }
        else {
            runNewObservable();
        }
        runSubscribe();
        return mainView;
    }

....

    private void runNewObservable () {
        mObservable =
                Observable.create(new Observable.OnSubscribe<Bitmap>() {
            @Override
            public void call(Subscriber<? super Bitmap> subscriber) {
                subscriber.onNext(new LoadingImage().loadImageFrom(imageURL));
                subscriber.onCompleted();
            }
        })
                .subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .cache();

    }



    private void runNewSubscribe () {
        mSubscriber = new Subscriber<Bitmap>() {

            @Override
            public void onCompleted() { }

            @Override
            public void onError(Throwable e) { }

            @Override
            public void onNext(Bitmap bitmap) {
                loadedBitmap = bitmap;
                imageView.setImageBitmap(bitmap);
            }
        };
    }

    @Override
    public void onResume() {
        super.onResume();
        mSubscription = mObservable.subscribe(mSubscriber);
    }


    @Override
    public void onSaveInstanceState(Bundle outState) {
        super.onSaveInstanceState(outState);
        outState.putParcelable("LoadedBitmap", loadedBitmap);
    }



    @Override
    public void onStop() {
        super.onStop();
        mSubscription.unsubscribe();
    }

....

如您所见。我只使用 mSubscription.subscribe() 和 mSubscription.unsubscribe() 并仅在 saveInstanceState 为 null 时创建 Observable。

Answer 2

查看rx.subjects.BehaviorSubject<T>。如果你不熟悉rx.subjects.Subjects，我能想到的描述它们的最一般的方式是它们破坏了 A 点和 B 点之间订阅的连续性。既是 Observer<T> 又如何呢？可以接受来自多个来源的onNext()s（警告：需要外螺纹安全）。另一方面，主题也是Observable<T>，因此可以订阅多个Observer<T>s，并且进来的onNext()s 将被多播到每个下游Observer<T>。

如果你的代码看起来像

Observable<T> src = ...;
Subscriber<T> dst;
src.subscribe(dst);

BehaviorSubject的使用方式是

Observable<T> src = ...;
BehaviorSubject<T> subject = BehaviorSubject.create(defaultValue);
src.subscribe(subject);

立即订阅源，主题将与发出的速度一样快。 BehaviorSubject 只保留最近的值，并删除 defaultValue 和所有以前的值。

// safe to do multiple times.
Subscriber<T> dst;
subject.subscribe(dst);

订阅时dst 在订阅后立即从src（或defaultValue）接收最新值，然后在dst 取消订阅之前的所有后续值。

警告：主题有过度使用的倾向，因此请确保您需要一个。