java - 如何阻止 rx-java runnable 发出？

Question

我有以下观察结果:

      ScheduledExecutorService executorService = Executors.newScheduledThreadPool( 1 );
      Observable<List<Widget>> findWidgetsObservable = Observable.create( emitter -> {
         executorService.scheduleWithFixedDelay( emitFindWidgets( emitter, 0, 30, TimeUnit.SECONDS );
      } );

     private Runnable emitFindWidgets( ObservableEmitter<List<Widgets>> emitter ) {
        return () -> {
              emitter.onNext( Collections.emptyList() ); // dummy empty array
        };
     }

我将它返回到 graphql-java 订阅解析器中，如下所示:

      ConnectableObservable<List<Widget>> connectableObservable = findWidgetsObservable.share().publish();
      Disposable connectionDisposable = connectableObservable.connect();
      return connectableObservable.toFlowable( BackpressureStrategy.LATEST )

graphql 订阅按预期工作并向 JavaScript graphql 客户端发送数据，但是当客户端取消订阅时，我的 Runnable 似乎无限地继续。也就是说，flowable 的 doOnCancel() 事件处理程序正在运行。

为了解决这个问题，我尝试在 flowable 的 doOnCancel() 中执行以下操作:

    Disposable connectionDisposable = connectableObservable.connect();
    return connectableObservable.toFlowable( BackpressureStrategy.LATEST ).doOnCancel( () -> {
             findWidgetsObservable.toFuture().cancel( true );
             connectionDisposable.dispose();
    })

但是，Runnable 会继续无限期地省略。有什么办法可以解决这个问题并完全停止发射？

我确实有一个想法:scheduleWithFixedDelay 返回一个 ScheduledFuture，它有一个 cancel() 方法，但我不确定当调度本身的范围在可观察范围内时我是否可以做到这一点!如有任何帮助，我们将不胜感激。

Answer 1

可运行对象继续发出，因为您正在调度器上安排发射，而该调度程序未知/绑定(bind)到可观察流。

当您处置连接时，您将停止从上游接收项目，因为与上游可观察对象的连接被切断。但由于您计划发射器在单独的调度程序上重复运行，因此可运行对象将继续运行。

您可以使用自定义调度程序来描述自定义调度行为，并将其传递到 subscribeOn(Your-Custom-Scheduler)

此外，您提到您可以在 中的 ScheduledFuture 上调用 cancel() doOnDispose()。

但是您应该在可观察链中显式切换调度程序。否则，调试会变得更加困难。