android reactivex主题线程

Question

我的聊天应用使用的是reactivex android。我有以下代码片段：

class ServerHelper {
  public static void requestNewMessages(Time after) {
    Requester<Message> req = new Requester<>(after) {
      public void onSuccess(List<Message> msgs) {
         // I think it's on main thread now, right? 
         DBHelper.storeMessages(msgs);
      }
    }
  }
}

class DBHelper {
  public static PublishSubject<List<MessageData>> subject = PublishSubject<>.create();
  public static void storeMessages(List<Message> msgs) {

    // Now do I need to switch to io thread? How

    List<MessageData> msgDataList = new ArrayList<>();
    for (Message msg : msgs) {
      MessageData msgData = msg.saveToDB();
      msgDataList.add(msgData);
    }

    // Now do I need to switch back to main thread? 

    subject.onNext(msgDataList);
  }
}

class MyFragment implements Observer<List<MessageData>> {
  private MessageData newestMessage;
  private RecyclerView recyclerView;
  public void onReceivedNewMessageNotification() {
     // it's ok for us to be on main thread here, right? since the call is async
     ServerHelper.requestNewMessages(newestMessage.time);
  }
  public void onResume() {
    super.onResume();
    DBHelper.subject.subscribe(this);
  }
  public void onNext(List<Message> data) {
    // is it on main thread now? 
    recyclerViewAdapter.setData(data);
    recyclerView.reloadData();
  }
}

基本上，我使用观察者模式来制作一个聊天应用程序。当用户收到新消息通知时，我想通过ServerHelper发送服务器请求，在成功回调中，我调用DBHelper的storeMessage，它通知主题。

我在代码中留下了一些注释。我很担心线程切换的问题。是否需要在storeMessages调用开始时切换到io线程，io完成后再切换回main线程？

另一个问题是，我在这里使用subject还是observable？哪一个更好？我知道使用subject，我们可以直接调用subject.onNext(msgs)，但似乎对于observables，onNext调用需要在observable定义中。

Answer 1

我想您可能会感到困惑，或者没有意识到您在Java中可以选择线程。您可以将任务或工作线程封装到一个对象(如Runnable)中，并将其推送到另一个线程中，而不是像您所说的那样全局更改程序的线程。看看Java以high-level concurrency的方式原生提供了什么。

就必要性而言，您也不需要切换到io线程。如果有一个主要的性能问题阻塞了数据库写操作的主线程，那么可以通过所有的方法。考虑:这样的线程决策也可能取决于你的错误处理。例如，写入数据库失败的消息是否仍会显示在msgDataList中？msgDataList是乐观地发送给观察者，还是等待数据库完成写入？

最后，为了解决Subject与Observable的比较问题:由于消息服务器处于可观察环境的最边缘，因此您或服务器库中离原始数据流非常近的某个部分必须使用Subject将项提供到Observable链中。在您的例子中，您可以认为它仅仅是随着时间的推移用Requester接口的onNext填充Observable接口的Observablees，但是您不能使用获取原始数据并在另一端充当Observable的东西。