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

我正在尝试迁移AsyncTask向服务器发送消息，使用 RxJava。粗略地说，该任务执行以下操作：

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

我已经创建了所需的 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));

当我订阅上述内容时，我会得到一个Disposable。通常我会把它添加到CompositeDisposable对象和clear该对象然后用户已移动到不同的view (i.e. fragment）。但是，在这种情况下，我需要继续运行此任务，以确保本地数据库相应地更新为任务结果。

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

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

一个处置Disposable一旦他不再对此感兴趣。

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

这将导致一种情况，当你Activity无法被垃圾收集，因为您的隐式引用Subscription，因此您正在创建内存泄漏情况。

如果您有这样的用例，我认为您必须在组件上执行该请求，该组件将独立于活动生命周期，例如Service https://developer.android.com/guide/components/services.html.