RxJava2 如何使工作线程在Disposable.dispose后完成流程

  在使用Rxjava2时，特别要注意内存泄漏。所以一般在activity或者fragment销毁时调用disposable.dispose来取消订阅。但在之前的工作中遇到一个问题，如何确保你在work thread中的流程走完。用过的同学会知道dispose会产生Exception导致线程中断。

 一种方式就是不在activity或者fragment销毁时调用disposable.dispose，在工作线程完成后再说。这样本人感觉并不好，因为我们一般在Consumer里对UI线程做更新，如：

new Consumer<Long>() {
@Override
public void accept(Long data) {
textView.setText("get data : " + data);
}
};

activity销毁时其实已经没有必要再更新UI，而且如果不释放的话会导致内存泄漏。所以要在取消订阅的情况下，继续完成工作线程里的流程（如对db的操作等），就要另想办法。

    查找一番后，发现一种方式比较不错，使用ReplaySubject来完成任务，先上代码：

ReplaySubject<Long> replaySubject;
Consumer<Long> consumer;
Disposable disposable;

@Override
protected void onCreate(Bundle savedInstanceState) {

......

replaySubject = ReplaySubject.create();

observable = Observable.just(123L)
.map(new Function<Long, Long>() {
@Override
public Long apply(Long data) {
Log.d("tag", ">>>>>1>>>>>   data=" + data);
try {
// emulate long time operation
Thread.sleep(15000);
Log.d("tag", ">>>>>2>>>>>   data=" + data);
} catch (InterruptedException e) {
e.printStackTrace();
}
return data;
}
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread());

consumer = new Consumer<Long>() {
@Override
public void accept(Long data) {
Log.d("tag", ">>>>>3>>>>>   data=" + data);
textView.setText("get data : " + data);
}
};

observable.subscribe(replaySubject);
disposable = replaySubject.subscribe(consumer);
}

@Override
protected void onDestroy() {

......

Log.d("tag", "~~~~~~~~~~~~~");
disposable.dispose();
consumer = null;
 }

这样在activity退出时ReplaySubject依然订阅着，使得工作线程继续完成任务，而Consumer 则可以消除以免内存泄漏。

当然如果大家有更好的方法，希望不吝指教。谢谢

阅读更多