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

我试图迁移将消息发送到服务器的AsyncTask，以使用RxJava。粗略地说，该任务执行以下操作：使用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));

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

最适合处理这种情况的方法是什么？我可以根本不将Disposable添加到我的CompositeDisposable对象中，因此它不会退订，但感觉它可能会导致问题。

P.S.通过观察SQLite表中的数据来处理对用户的更新。这些事件由synchroniseMessage方法触发。这是一个不同的订阅，我只是退订，所以它不是问题的一部分。

来源

2017-04-21 vkislicins

一旦他不再对它感兴趣，就会立即处置Disposable。

在您的情况下，您仍然对流感兴趣，无论用户导航到其他屏幕还是不显示，这意味着您无法取消订阅。这意味着你不能将它添加到CompositeDisposable。

这会导致一种情况，当您的Activity不能被垃圾收集时，由于您的Subscription中隐含引用它，因此您正在创建内存泄漏情况。

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

来源

2017-04-21 15:14:07 azizbekian

我真的想避免使用'服务'或回到使用'AsyncTask'。这既会对我的架构产生影响，也会减少我对单元测试逻辑的选择。我试图了解如何不保留一个'Disposable'的引用，导致它最终会终止（15秒'OkHttp'超时期）并处理订阅......或者我可以存储该订阅在某处可以访问，并有一个轮询任务，以便每隔一段时间一次性提取一次... – vkislicins

我看到的唯一问题是，您会阻止GC回收您的“活动”拥有的内存，持续15秒。之后，下一个GC事件通常会清理活动。所以，基本上，你只有15秒的表现，就像一个坏男孩。 – azizbekian

但是，如果用户连续多次保持旋转设备，并且您每次都重新订阅 - 将发生“OutOfMemoryException”。 – azizbekian

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

回答

相关问题