使用RxJava处理长时间运行的任务

Question

我正在尝试将一个向服务器发送消息的异步任务迁移到RxJava。该任务大致执行以下操作：

1） 创建将要发送的消息（保存到数据库）
2）向用户显示消息（状态为“发送”）
3）将消息发送到服务器（下面的代码段）
4）将消息标记为已发送或失败（保存到数据库）
5）更新UI

我创建了所需的Rx链，部分如下所示：

 public Observable<Message> sendMessage(Message message) {
     return mApiClient.sendMessage(message)
         .doOnNext(sentMessage -> mDatabase.synchroniseMessage(sentMessage))
         .doOnError(e -> {
             message.setState(FAILED);
             mDatabase.synchroniseMessage(message));
         })
         .onErrorReturn(e -> Observable.just(message));

当我订阅上述内容时，我会得到一个一次性的。通常，我会将其添加到CompositeDisposable对象中，然后清除该对象，然后用户移动到不同的视图（即片段）。但是，在这种情况下，我需要继续运行此任务，以确保本地数据库使用相应的任务结果进行更新。

处理这种情况的最合适方法是什么？我无法简单地将Disposable添加到我的CompositeDisposable对象中，因此它不会被取消订阅，但感觉它可能会导致问题。

附言。向用户显示更新是通过观察SQLite表中的数据来处理的。这些事件是由syniseMessage方法触发的。这是一个不同的订阅，我将简单地取消订阅，所以这不是问题的一部分。

Answer 1

一旦他不再对Disposable感兴趣，就会处理掉它。

在您的情况下，无论用户是否导航到另一个屏幕，您仍然对该流感兴趣，这意味着您无法取消订阅。这意味着您无法将其添加到CompositeDisposable。

这将导致一种情况，当您的活动无法被垃圾收集时，因为您的订阅隐式引用了它，因此您正在创建内存泄漏情况。

如果您有这样一个用例，我认为您必须对一个组件执行该请求，该组件将独立于活动生命周期，如服务。