使用OkHttp3和ReactiveX Java实现长轮询的正确方法

Question

下面的代码演示了我正在尝试做的事情（似乎有效），但我很确定这是不行的。

// return Observable<Response>
makeGetObservable("http://my.service.com/api/events")
  // check for error and map to Observable<ResponseBody>
  .map(this::mapRespBodyOrError)
  // flat map to Observable<String> representing line of long polling response
  .flatMap(respBody -> Observable.create(emitter -> {
    // open reader on response body stream
    try (BufferedReader reader = new BufferedReader(respBody.charStream())) {
      String line;
      // block and wait to read a line from input
      while((line = reader.readLine()) !=null) {
        // once line was read from response body input stream emit it as observable event
        emitter.onNext(line);
      }
    }
  }));

Answer 1

经过一些研究，我发现被阻塞的线程是OkHttp客户端线程。这是由于我实现了MakegetObservable，它从OkHttp客户机的newcall...enqueue回调发出。默认情况下，OkHttp客户端有5个线程池，用于一个资源的5个并发连接。因此，每次订阅另一个长轮询资源时，我都会阻塞其中一个线程。在5个阻塞线程之后，OkHttp客户端停止工作，因为它没有线程来处理响应。

正如@Progman所建议的，我使用subscribeon使用IO调度器，它为每个阻塞IO操作生成新线程。一个必须小心和适当地处置资源与这个调度器。

我的实现目前如下所示（添加了调度程序、完成和错误事件）

// return Observable<Response>
makeGetObservable("http://my.service.com/api/events")
  // check for error and map to Observable<ResponseBody>
  .map(this::mapRespBodyOrError)
  // use IO scheduler that spawns new threads to take care of blocking operations
  .observeOn(Schedulers.io())
  // flat map to Observable<String> representing line of long polling response
  .flatMap(respBody -> Observable.create(emitter -> {
    try (BufferedReader reader = new BufferedReader(respBody.charStream())) {
      // blocking read lines while available
      String line;
      while ((line = reader.readLine()) != null) {
        // emit event for every line
        emitter.onNext(line);
      }
      // emit the completion event to indicate we are done
      emitter.onComplete();
    } catch (IOException | RuntimeException err) {
      // emit any error that might have occurred
      emitter.onError(err);
    }
  }));