背景

对于金融交易软件，行情信息的实时更新是一个重要需求。后端团队已经提供了 WebSocket api，现在需求在安卓客户端使用 WebSocket 来接收后端实时推送的行情更新。

平台

Android

语言环境

Kolin

目标

• 在异步线程里建立 WebSocket 连接
• WebSocket 所有事件回调到主线程处理
• 支持消息背压（行情可能在短时间内有剧烈波动，消息量会激增）

实现

建立连接

WebSocket 采用了 okHttp 的内置实现 - RealWebSocket，其中 WebSocketListener 提供了相应事件回调:

public abstract class WebSocketListener {
  /**
   * Invoked when a web socket has been accepted by the remote peer and may begin transmitting
   * messages.
   */
  public void onOpen(WebSocket webSocket, Response response) {
  }

  /** Invoked when a text (type {@code 0x1}) message has been received. */
  public void onMessage(WebSocket webSocket, String text) {
  }

  /** Invoked when a binary (type {@code 0x2}) message has been received. */
  public void onMessage(WebSocket webSocket, ByteString bytes) {
  }

  /**
   * Invoked when the remote peer has indicated that no more incoming messages will be
   * transmitted.
   */
  public void onClosing(WebSocket webSocket, int code, String reason) {
  }

  /**
   * Invoked when both peers have indicated that no more messages will be transmitted and the
   * connection has been successfully released. No further calls to this listener will be made.
   */
  public void onClosed(WebSocket webSocket, int code, String reason) {
  }

  /**
   * Invoked when a web socket has been closed due to an error reading from or writing to the
   * network. Both outgoing and incoming messages may have been lost. No further calls to this
   * listener will be made.
   */
  public void onFailure(WebSocket webSocket, Throwable t, @Nullable Response response) {
  }
}

这里有两个地方需要注意，onClosed 和 onFailure，当两个对等方都指示不再传输消息并且连接已成功释放时，onClosed 会被调用，之后不会再有任何回调，而由于从网络读取或写入错误而关闭 WebSocket 时，onFailure 会被调用，同样之后不会再有任何回调，注意此时 WebSocket 会被关闭而且 onClosed 不会再被调用，这是 RealWebSocket 的实现：

public final class RealWebSocket implements WebSocket, WebSocketReader.FrameCallback {
  ...
  public void failWebSocket(Exception e, @Nullable Response response) {
    Streams streamsToClose;
    synchronized (this) {
      if (failed) return; // Already failed.
      failed = true;
      streamsToClose = this.streams;
      this.streams = null;
      if (cancelFuture != null) cancelFuture.cancel(false);
      if (executor != null) executor.shutdown(); //停止 executor
    }

    try {
      listener.onFailure(this, e, response);
    } finally {
      //关闭流
      closeQuietly(streamsToClose);
    }
  }
  ...
}

另外双方还需要一个心跳机制来检测连接状态，RealWebSocket 内心跳是一个空字节，即 0 byte，心跳逻辑与后端 api 一致，因此心跳不做修改，如果实际业务中心跳机制不同于此，就需要做修改了，这里暂且跳过。

下面我们来建立一个 WebSocket 连接（部分代码省略）：

val okHttpClient = OkHttpClient.Builder()..........build()
val request = Request.Builder().url("这里输入连接地址").build()
val listener = object: WebSocketListener () {
        ...
}
val pingIntervalMillis = 1000 * 60 * 5 //心跳间隔 5 分钟
val realWss = RealWebSocket(request, listener, SecureRandom(), pingIntervalMillis)
realWss.connect(client) //建立连接

这样我们就能建立起一个 WebSocket 连接了，所有消息都会回调到我们定义的 listener 里。

异步回调

大家都知道 Android 网络操作一定不能放在主线程中，这一点就不做赘述了。这里我们使用了 RxJava2 的 Flowable 操作符以及 Scheduler 来进行线程调度，我们可以手动调用 onNext 来向订阅者发送 WebSocket 消息事件，同时 Flowable 也支持背压：

public enum BackpressureStrategy {
    /**
     * OnNext events are written without any buffering or dropping.
     * Downstream has to deal with any overflow.
     * <p>Useful when one applies one of the custom-parameter onBackpressureXXX operators.
     */
    MISSING,
    /**
     * Signals a MissingBackpressureException in case the downstream can't keep up.
     */
    ERROR,
    /**
     * Buffers <em>all</em> onNext values until the downstream consumes it.
     */
    BUFFER,
    /**
     * Drops the most recent onNext value if the downstream can't keep up.
     */
    DROP,
    /**
     * Keeps only the latest onNext value, overwriting any previous value if the
     * downstream can't keep up.
     */
    LATEST
}

以上是 RxJava2 提供的几种背压策略，根据业务这里选择了 BUFFER 缓存所有收到的消息直到被消费。同时 BUFFER 也是默认的背压策略：

public final class FlowableCreate<T> extends Flowable<T> {
    ...
    @Override
    public void subscribeActual(Subscriber<? super T> t) {
        BaseEmitter<T> emitter;

        switch (backpressure) {
        ...
          default: {
            //默认 BufferAsyncEmitter
            emitter = new BufferAsyncEmitter<T>(t, bufferSize());
            break;
          }
        }
        ...
    }
    ...
}

这里我们注意到，BufferAsyncEmitter 实例化的过程中传入了 bufferSize() ，通过查看 Flowable 我们看到默认的 buffer size 是 128：

public abstract class Flowable<T> implements Publisher<T> {
        /** The default buffer size. */
    static final int BUFFER_SIZE;
    static {
        BUFFER_SIZE = Math.max(1, Integer.getInteger("rx2.buffer-size", 128));
    }
    ...
}

这里先使用默认容量，不做修改。

接下来将 WebSocket 所有事件密封在一个类里，用作 Flowable 的消息类型：

sealed class WebSocketEvent {

    class Open(webSocket: WebSocket?, val response: Response?) : WebSocketEvent()

    class BinaryMessage(webSocket: WebSocket?, val bytes: ByteString?) : WebSocketEvent()

    class StringMessage(webSocket: WebSocket?, val text: String?) : WebSocketEvent()

    class Closing(webSocket: WebSocket?, val code: Int, val reason: String?) : WebSocketEvent()

    class Closed(webSocket: WebSocket?, val code: Int, val reason: String?) : WebSocketEvent()


}

在 WebSocketListener 中向订阅者发送对应的事件：

class FlowableWebSocketListener(private val emitter: FlowableEmitter<WebSocketEvent>) : WebSocketListener() {

    override fun onOpen(webSocket: WebSocket?, response: Response?) {
        emitter.onNext(WebSocketEvent.Open(webSocket, response))
    }

    override fun onMessage(webSocket: WebSocket?, text: String?) {
        emitter.onNext(WebSocketEvent.StringMessage(webSocket, text))
    }

    override fun onMessage(webSocket: WebSocket?, bytes: ByteString?) {
        emitter.onNext(WebSocketEvent.BinaryMessage(webSocket, bytes))
    }

    override fun onClosing(webSocket: WebSocket?, code: Int, reason: String?) {
        emitter.onNext(WebSocketEvent.Closing(webSocket, code, reason))
    }

    override fun onClosed(webSocket: WebSocket?, code: Int, reason: String?) {
        emitter.onComplete()
    }

    override fun onFailure(webSocket: WebSocket?, t: Throwable?, response: Response?) {
        if (!emitter.isCancelled) {
            emitter.onError(t ?: IOException("WebSocket unknown error"))
            emitter.onComplete()
        }
    }
}

接下来我们创建一个 Flowable，在异步线程建立连接，消息发送到主线程做处理：

val flowable: Flowable<WebSocketEvent> = Flowable.create({ emitter ->
      ...
      val listener = RxWebSocketListener(emitter)
      val realWss = RealWebSocket(request, listener, SecureRandom(), pingIntervalMillis)
      realWss.connect(client) //建立连接
}, BackpressureStrategy.BUFFER)

flowable
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(object: ResourceSubscriber<WebSocketEvent> {
      ...
      override fun onStart() {
         //ResourceSubscriber 会一次拉取 Long.MAX_VALUE 个消息，这里我们覆写先拉取一条
         request(1)
      }

      override fun onNext(t: WebSocketEvent?) {
          when(t){
              WebSocketEvent.Open -> {...}
              WebSocketEvent.BinaryMessage -> {...}
              WebSocketEvent.StringMessage -> {...}
              WebSocketEvent.Closing -> {...}
          }

          if (判断是否需要继续拉取消息) {
              request(1)
          }
      }

      override fun onComplete() {
           onClosed()
           dispose()
      }
      ...
})

这里要注意 Schedulers.io() 创建工作线程的数量是没有上限的，因此在 WebSocket 关闭之后应该立即 dispose() 来释放，否则可能引发 OutOfMemory。

...
static final class CachedWorkerPool implements Runnable {
    ThreadWorker get() {
          if (allWorkers.isDisposed()) {
              return SHUTDOWN_THREAD_WORKER;
          }
          while (!expiringWorkerQueue.isEmpty()) {
              ThreadWorker threadWorker = expiringWorkerQueue.poll();
              if (threadWorker != null) {
                  return threadWorker;
              }
          }

          // No cached worker found, so create a new one.
          // 缓存中没有可复用的 ThreadWorker 时，创建一个新的 ThreadWorker
          ThreadWorker w = new ThreadWorker(threadFactory);
          allWorkers.add(w);
          return w;
    }
    ...
}
...

到此一个异步 WebSocket 连接就搭建完成了。

（转载请注明出处）