获取对原始调度器的引用

Question

我有以下 RxJava Observable:

final class MapBitmapObservable {

    static Observable<Bitmap> create(@NonNull final MapView mapView) {
        return Observable.create(new Observable.OnSubscribe<Bitmap>() {
            @Override
            public void call(final Subscriber<? super Bitmap> subscriber) {
                mapView.getMapAsync(new OnMapReadyCallback() {
                    @Override
                    public void onMapReady(@NonNull final GoogleMap googleMap) {
                        googleMap.snapshot(new GoogleMap.SnapshotReadyCallback() {
                            @Override
                            public void onSnapshotReady(@Nullable final Bitmap bitmap) {
                                if (bitmap != null) {
                                    subscriber.onNext(bitmap);
                                    subscriber.onCompleted();
                                } else {
                                    subscriber.onError(new MapSnapshotFailedException());
                                }
                            }
                        });
                    }
                });
            }
        });
    }

    private MapBitmapObservable() {

    }

}

必须在主线程上调用MapView方法getMapAsync来避免这个异常：

java.lang.IllegalStateException: getMapAsync() must be called on the main thread
    at com.google.android.gms.common.internal.zzx.zzcD(Unknown Source)
    at com.google.android.gms.maps.MapView.getMapAsync(Unknown Source)
    at com.github.stkent.bugshaker.email.screenshot.maps.MapBitmapObservable$1.call(MapBitmapObservable.java:42)
    at com.github.stkent.bugshaker.email.screenshot.maps.MapBitmapObservable$1.call(MapBitmapObservable.java:37)
    at rx.Observable.unsafeSubscribe(Observable.java:8098)
    ...

假设MapBitmapObservable 被用作Observable 链的一部分，其中先前和后续操作可能会长时间运行，并且应该在主线程之外执行。一个简化的示例可能如下所示：

Observable.just(activity)
        .flatmap(new Func1<Activity, Observable<MapView>>() {
            @Override
            public Observable<Bitmap> call(@NonNull final Activity activity) {
                return ExpensiveToCreateObservable.create(activity);
            }
        })
        .flatmap(new Func1<MapView, Observable<Bitmap>>() {
            @Override
            public Observable<Bitmap> call(@NonNull final MapView mapView) {
                return MapBitmapObservable.create(mapView);
            }
        })
        .flatmap(new Func1<Bitmap, Observable<Uri>>() {
            @Override
            public Observable<Uri> call(@NonNull final Bitmap bitmap) {
                return SomeOtherExpensiveToCreateObservable.create(bitmap);
            }
        })
        .subscribeOn(Schedulers.io())
        .subscribe();

（虽然应该注意，在我的实际应用中，链接分布在几种不同的方法中）。我想：

1. 确保在主线程上调用MapView.getMapAsync；
2. 允许在原始调度程序上执行第二个长时间运行的操作，无论是什么（Schedulers.io()、Schedulers.computation() 等）李>

在我看来，实现此目的的伪代码如下所示：

Observable.just(activity)
        .flatmap(new Func1<Activity, Observable<MapView>>() {
            @Override
            public Observable<Bitmap> call(@NonNull final Activity activity) {
                return ExpensiveToCreateObservable.create(activity);
            }
        })
        .observeOn(AndroidSchedulers.mainThread()) // This is real, and resolves bullet 1.
        .flatmap(new Func1<MapView, Observable<Bitmap>>() {
            @Override
            public Observable<Bitmap> call(@NonNull final MapView mapView) {
                return MapBitmapObservable.create(mapView);
            }
        })
        .observeOn(/* Some way of referencing the thread on which I originally subscribed, to resolve bullet 2. */)
        .flatmap(new Func1<Bitmap, Observable<Uri>>() {
            @Override
            public Observable<Uri> call(@NonNull final Bitmap bitmap) {
                return SomeOtherExpensiveToCreateObservable.create(bitmap);
            }
        })
        .subscribeOn(Schedulers.io()) // I do not want to rely on knowledge of the Scheduler type used at this call-site. 
        .subscribe();

这可能吗？

Answer 1

来自observeOn() documentation:

另一方面，ObserveOn 会影响 Observable 将在该运算符出现的下方使用的线程。因此，您可以在 Observable 运算符链中的不同点多次调用 ObserveOn，以更改某些运算符在哪些线程上运行。

正如 Aaron He 所提到的，您可以保留一些对您正在使用的调度程序的引用，在后者“observeOn”上使用它。

---

我有时使用的另一种方法是删除“observeOn”函数，并确保 Activity.runOnUiThread 在 UI 线程上处理 View 项目。类似的东西 -

static Observable<Bitmap> create(@NonNull final Activity activity,@NonNull final SomeObject someObject) {
        return Observable.create(new Observable.OnSubscribe<Pair<Activity,SomeObject>>() {
            @Override
            public void call(final Subscriber<? super Pair<Activity,SomeObject>> subscriber) {
                activity.runOnUiThread(new Runnable() {
                    @Override
                    public void run() {
                        someObject.doStuff();
                    }
                });  
            }
        });
    }