为什么我的线程计数会因 rxjava 重试错误而增加?

Question

在函数方法中，我抛出一个nullPointerException并打印出线程计数(sout(Thread.activeCount(。似乎每次我重试错误时，线程计数都会增加 1。为什么会这样？我应该假设它只会被垃圾收集吗？

public void start(int time) {
Observable.interval(time, TimeUnit.SECONDS)
.doOnNext(t -> function())
.doOnError(System.out::println)
.retry()
.subscribe();

}

Answer 1

默认情况下，该interval过载使用的computationScheduler仅限于 CPU 的数量(Runtime.getRuntime().availableProcessors()(。当您由于重试而不断重新订阅时，Scheduler将启动越来越多的工作线程。但是，总线程数不应无休止地增长。

如果您不想有这么多工作线程，请考虑使用Schedulers.single()或Schedulers.io()后者重用工作线程：

Observable.interval(time, TimeUnit.SECONDS, Schedulers.single())
.doOnNext(t -> function())
.doOnError(System.out::println)
.retry()
.subscribe();

或

Observable.interval(time, TimeUnit.SECONDS, Schedulers.io())
.doOnNext(t -> function())
.doOnError(System.out::println)
.retry()
.subscribe();

Answer 2

也许每次重试时，Observable.interval(( 都会从其计算调度程序线程池中获取新线程。也许这样做是因为当时以前的线程仍在使用中？恕我直言，它不应该无限期地上升，因为它会指示一些错误。