java - RxAndroid 在相机更改时去抖谷歌地图

Question

我有一个使用谷歌地图并监听相机变化的应用程序。我的问题是每次更改相机时我都必须请求我的后端。我想做的只是通过使用 RxAndroid/Java 去抖动来限制请求的数量。

我的代码是这样的:

Observable.create(new Observable.OnSubscribe<CameraPosition>() {
        @Override
        public void call(Subscriber<? super CameraPosition> subscriber) {
            if (!subscriber.isUnsubscribed()) {
                map.setOnCameraChangeListener(new GoogleMap.OnCameraChangeListener() {
                    @Override
                    public void onCameraChange(CameraPosition cameraPosition) {
                        subscriber.onNext(cameraPosition);
                    }
                });
            }
        }
    }).subscribeOn(AndroidSchedulers.mainThread())
            .observeOn(AndroidSchedulers.mainThread())
            .onErrorResumeNext(Observable.<CameraPosition>empty())
            .debounce(1, TimeUnit.SECONDS)
            .subscribe(cameraPosition -> {
                final LatLngBounds item = map.getProjection().getVisibleRegion().latLngBounds;

                homeActionBarActivity.getNMB().getRide().list(
                        item.southwest.latitude,
                        item.southwest.longitude,
                        item.northeast.latitude,
                        item.northeast.longitude)
                        .subscribeOn(Schedulers.newThread())
                        .observeOn(AndroidSchedulers.mainThread())
                        .onErrorResumeNext(Observable.<List<Ride>>empty())
                        .subscribe(new Action1<List<Ride>>() {
                            @Override
                            public void call(List<Ride> rides) {

                                clearRidePointsFromMap();

                                for (Ride ride : rides) {
                                    if (isAdded())
                                        addRideStartPointToMap(ride);
                                }

                            }
                        });
            });

如您所见，我强制对两者(subscribeOn 和 observeOn)都使用 MainThread，但我仍然收到此错误“不在主线程上”。

这里是堆栈跟踪:

01-27 10:59:24.049    3049-3066/com.nousmotards.android E/AndroidRuntime﹕ FATAL EXCEPTION: RxComputationThreadPool-2
    Process: com.nousmotards.android, PID: 3049
    java.lang.IllegalStateException: Exception thrown on Scheduler.Worker thread. Add `onError` handling.
            at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:52)
            at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:422)
            at java.util.concurrent.FutureTask.run(FutureTask.java:237)
            at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:152)
            at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:265)
            at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1112)
            at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:587)
            at java.lang.Thread.run(Thread.java:864)
     Caused by: rx.exceptions.OnErrorNotImplementedException: Not on the main thread
            at rx.Observable$31.onError(Observable.java:7134)
            at rx.observers.SafeSubscriber._onError(SafeSubscriber.java:154)
            at rx.observers.SafeSubscriber.onError(SafeSubscriber.java:111)
            at rx.observers.SafeSubscriber.onNext(SafeSubscriber.java:137)
            at rx.observers.SerializedObserver.onNext(SerializedObserver.java:159)
            at rx.observers.SerializedSubscriber.onNext(SerializedSubscriber.java:81)
            at rx.internal.operators.OperatorDebounceWithTime$DebounceState.emit(OperatorDebounceWithTime.java:128)
            at rx.internal.operators.OperatorDebounceWithTime$1$1.call(OperatorDebounceWithTime.java:72)
            at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:47)
            at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:422)
            at java.util.concurrent.FutureTask.run(FutureTask.java:237)
            at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:152)
            at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:265)
            at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1112)
            at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:587)
            at java.lang.Thread.run(Thread.java:864)
     Caused by: java.lang.IllegalStateException: Not on the main thread
            at com.google.k.a.cl.b(Unknown Source)
            at com.google.maps.api.android.lib6.c.ca.a(Unknown Source)
            at com.google.maps.api.android.lib6.c.el.l(Unknown Source)
            at com.google.android.gms.maps.internal.l.onTransact(SourceFile:312)
            at android.os.Binder.transact(Binder.java:361)
            at com.google.android.gms.maps.internal.IGoogleMapDelegate$a$a.getProjection(Unknown Source)
            at com.google.android.gms.maps.GoogleMap.getProjection(Unknown Source)
            at com.nousmotards.android.fragments.home.MapFragment.lambda$onViewCreated$33(MapFragment.java:117)
            at com.nousmotards.android.fragments.home.MapFragment.access$lambda$0(MapFragment.java)
            at com.nousmotards.android.fragments.home.MapFragment$$Lambda$1.call(Unknown Source)
            at rx.Observable$31.onNext(Observable.java:7139)
            at rx.observers.SafeSubscriber.onNext(SafeSubscriber.java:130)
            at rx.observers.SerializedObserver.onNext(SerializedObserver.java:159)
            at rx.observers.SerializedSubscriber.onNext(SerializedSubscriber.java:81)
            at rx.internal.operators.OperatorDebounceWithTime$DebounceState.emit(OperatorDebounceWithTime.java:128)
            at rx.internal.operators.OperatorDebounceWithTime$1$1.call(OperatorDebounceWithTime.java:72)
            at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:47)
            at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:422)
            at java.util.concurrent.FutureTask.run(FutureTask.java:237)
            at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:152)
            at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:265)
            at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1112)
            at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:587)
            at java.lang.Thread.run(Thread.java:864)

你有什么想法吗？

注意:如果我从 Observable.create(...) 中取出 map.setOnCameraChangeListerner(...) 它会正常工作。

Answer 1

好吧，这似乎(可能)是由于 RxJava 内部关于如何管理将对象从订阅者传递给观察者的原因。我通过简单地在订阅者而不是观察者中获取 LatLngBounds 来解决我的问题。

Observable.create(new Observable.OnSubscribe<LatLngBounds>() {
    @Override
    public void call(Subscriber<? super LatLngBounds> subscriber) {
        if (!subscriber.isUnsubscribed()) {
            map.setOnCameraChangeListener(cameraPosition ->
                    subscriber.onNext(map.getProjection().getVisibleRegion().latLngBounds));
        }
    }
}).subscribeOn(AndroidSchedulers.mainThread())
        .observeOn(AndroidSchedulers.mainThread())
        .onErrorResumeNext(Observable.<LatLngBounds>empty())
        .debounce(1, TimeUnit.SECONDS)
        .subscribe(item -> {
            homeActionBarActivity.getNMB().getRide().list(
                    item.southwest.latitude,
                    item.southwest.longitude,
                    item.northeast.latitude,
                    item.northeast.longitude)
                    .subscribeOn(Schedulers.newThread())
                    .observeOn(AndroidSchedulers.mainThread())
                    .onErrorResumeNext(Observable.<List<Ride>>empty())
                    .subscribe(new Action1<List<Ride>>() {
                        @Override
                        public void call(List<Ride> rides) {

                            clearRidePointsFromMap();

                            for (Ride ride : rides) {
                                if (isAdded())
                                    addRideStartPointToMap(ride);
                            }

                        }
                    });
        });

希望这会有所帮助:)