RX And​​roid Observable.interval() 导致OOM

Question

我可能应该从我是 android RX 世界的新手开始。我有一个轮询方法，我使用 Observable.interval() 每秒轮询一次：

Observable.interval(0, 1, TimeUnit.SECONDS)
                    .timeInterval()
                    .subscribeOn(Schedulers.io())
                    .subscribe(new Consumer<Timed<Long>>() {
                        @Override
                        public void accept(@NonNull Timed<Long> longTimed) throws Exception 
                            {
                              presenter.poll()
                            }

                        }
                    });

现在这可以正常工作一段时间，但在我遇到内存不足异常之后。我的猜测是我创建的线程非常快，GC 无法及时清除它们。

1638 W System.err:  at io.reactivex.plugins.RxJavaPlugins.onError(RxJavaPlugins.java:366)
04-09 09:19:55.634  1543  1638 W System.err:    at io.reactivex.internal.schedulers.ScheduledDirectPeriodicTask.run(ScheduledDirectPeriodicTask.java:42)
04-09 09:19:55.634  1543  1638 W System.err:    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:462)
04-09 09:19:55.634  1543  1638 W System.err:    at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:307)
04-09 09:19:55.634  1543  1638 W System.err:    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:302)
04-09 09:19:55.634  1543  1638 W System.err:    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1167)
04-09 09:19:55.634  1543  1638 W System.err:    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:641)
04-09 09:19:55.634  1543  1638 W System.err:    at java.lang.Thread.run(Thread.java:919)
04-09 09:19:55.634  1543  1638 W System.err: Caused by: java.lang.OutOfMemoryError: pthread_create (1040KB stack) failed: Try again
04-09 09:19:55.634  1543  1638 W System.err:    at java.lang.Thread.nativeCreate(Native Method)
04-09 09:19:55.634  1543  1638 W System.err:    at java.lang.Thread.start(Thread.java:883)
04-09 09:19:55.634  1543  1638 W System.err:    at java.util.concurrent.ThreadPoolExecutor.addWorker(ThreadPoolExecutor.java:975)
04-09 09:19:55.634  1543  1638 W System.err:    at java.util.concurrent.ThreadPoolExecutor.ensurePrestart(ThreadPoolExecutor.java:1617)
04-09 09:19:55.634  1543  1638 W System.err:    at java.util.concurrent.ScheduledThreadPoolExecutor.delayedExecute(ScheduledThreadPoolExecutor.java:342)
04-09 09:19:55.634  1543  1638 W System.err:    at java.util.concurrent.ScheduledThreadPoolExecutor.schedule(ScheduledThreadPoolExecutor.java:579)
04-09 09:19:55.634  1543  1638 W System.err:    at java.util.concurrent.ScheduledThreadPoolExecutor.submit(ScheduledThreadPoolExecutor.java:680)
04-09 09:19:55.634  1543  1638 W System.err:    at io.reactivex.internal.schedulers.NewThreadWorker.scheduleActual(NewThreadWorker.java:146)
04-09 09:19:55.634  1543  1638 W System.err:    at io.reactivex.internal.schedulers.IoScheduler$EventLoopWorker.schedule(IoScheduler.java:230)
04-09 09:19:55.634  1543  1638 W System.err:    at io.reactivex.Scheduler.scheduleDirect(Scheduler.java:136)
04-09 09:19:55.634  1543  1638 W System.err:    at io.reactivex.Scheduler.scheduleDirect(Scheduler.java:112)
04-09 09:19:55.635  1543  1638 W System.err:    at io.reactivex.internal.operators.single.SingleSubscribeOn.subscribeActual(SingleSubscribeOn.java:37)
04-09 09:19:55.635  1543  1638 W System.err:    at io.reactivex.Single.subscribe(Single.java:3096)
04-09 09:19:55.635  1543  1638 W System.err:    at mypackagename.Presenter.poll(Presenter.java:83)

有什么办法可以重用调度程序？还是在我投票后手动删除/关闭它？

Answer 1

声明一个 Disposable 类型的变量：这样如果我们需要停止它，就会使用它。

Disposable disposable;

然后按照下面的代码：

disposable = Observable.interval(1, TimeUnit.SECONDS)
                    .doOnNext(t -> YOUR_FUNCTION_NAME())
                    .subscribe();

当您需要停止时，只需使用以下代码：

  if (!disposable.isDisposed()) {
                disposable.dispose();

注意：将 YOUR_FUNCTION_NAME 替换为您的函数名称。