HTML5 擁有許多引人注目的新特性，WebSocket就是其中之一。WebSocket一向有著 “Web 的 TCP ”之稱。通常 WebSocket 都是用于Web的，用于構建實時的 Web 應用。它可以在瀏覽器和服務器之間提供一個基于 TCP 連接的雙向通道。

WebSocket 協議本質上是一個基于 TCP 的協議。為了建立一個 WebSocket 連接，客戶端瀏覽器首先要向服務器發起一個 HTTP 請求，這個請求和通常的 HTTP 請求不同，包含了一些附加頭信息，其中附加頭信息 ”Upgrade: WebSocket” 表明這是一個申請協議升級的 HTTP 請求，服務器端解析這些附加的頭信息然后產生應答信息返回給客戶端，客戶端和服務器端的 WebSocket 連接就建立起來了，雙方就可以通過這個連接通道自由的傳遞信息，并且這個連接會持續存在直到客戶端或者服務器端的某一方主動的關閉連接。

Websocket同樣可以用于移動端。盡管移動端 Android/iOS 的本地應用可以直接通過Socket與服務器建立連接，并定義自己的協議來解決 Web 中實時應用創建困難的問題，但 WebSocket 服務通常復用Web的 80 端口，且可以比較方便的基于Web服務器來實現，因而對于某些端口容易被封的網絡環境而言，WebSocket 就變得非常有意義。

OkHttp 是在 2016 年 6 月 10 日發布的 3.4.1 版中添加的對WebSocket的支持的。本文通過分析 OkHttp-3.5.0 的 WebSocket 實現來學習一下這個協議。

OkHttp WebSocket客戶端 API 用法

在開始分析 WebSocket 的實現之前，我們先來看一下 OkHttp 的 WebSocket API怎么用。示例代碼如下：

import android.util.Log;import okhttp3.OkHttpClient; import okhttp3.Request; import okhttp3.Response; import okhttp3.WebSocket; import okhttp3.WebSocketListener; import okio.ByteString;public class WebsocketClient {private static final int NORMAL_CLOSURE_STATUS = 1000;private static OkHttpClient sClient;private static WebSocket sWebSocket;public static synchronized void startRequest() {if (sClient == null) {sClient = new OkHttpClient();}if (sWebSocket == null) {Request request = new Request.Builder().url("ws://echo.websocket.org").build();EchoWebSocketListener listener = new EchoWebSocketListener();sWebSocket = sClient.newWebSocket(request, listener);}}private static void sendMessage(WebSocket webSocket) {webSocket.send("Knock, knock!");webSocket.send("Hello!");webSocket.send(ByteString.decodeHex("deadbeef"));}public static void sendMessage() {WebSocket webSocket;synchronized (WebsocketClient.class) {webSocket = sWebSocket;}if (webSocket != null) {sendMessage(webSocket);}}public static synchronized void closeWebSocket() {if (sWebSocket != null) {sWebSocket.close(NORMAL_CLOSURE_STATUS, "Goodbye!");sWebSocket = null;}}public static synchronized void destroy() {if (sClient != null) {sClient.dispatcher().executorService().shutdown();sClient = null;}}private static void resetWebSocket() {synchronized (WebsocketClient.class) {sWebSocket = null;}}public static class EchoWebSocketListener extends WebSocketListener {private static final String TAG = "EchoWebSocketListener";@Overridepublic void onOpen(WebSocket webSocket, Response response) {sendMessage(webSocket);}@Overridepublic void onMessage(WebSocket webSocket, String text) {Log.i(TAG, "Receiving: " + text);}@Overridepublic void onMessage(WebSocket webSocket, ByteString bytes) {Log.i(TAG, "Receiving: " + bytes.hex());}@Overridepublic void onClosing(WebSocket webSocket, int code, String reason) {webSocket.close(NORMAL_CLOSURE_STATUS, null);Log.i(TAG, "Closing: " + code + " " + reason);resetWebSocket();}@Overridepublic void onClosed(WebSocket webSocket, int code, String reason) {Log.i(TAG, "Closed: " + code + " " + reason);}@Overridepublic void onFailure(WebSocket webSocket, Throwable t, Response response) {t.printStackTrace();resetWebSocket();}} }

這個過程與發送HTTP請求的過程有許多相似之處，它們都需要創建 OkHttpClient 和Request。然而它們不同的地方更多：

WebSocket 請求通過 WebSocketListener 來接收連接的狀態和活動，而HTTP請求則通過 Callback。同時請求的 URL 的 scheme 是 “ws” 或者是 “wss” (TLS 之上的 WebSocket)，而不是HTTP的 "http" 和 "https"。

HTTP 請求的連接建立及執行需要基于 Request 和回調創建Call，并調用 Call 的方法手動進行；而對于 WebSocket 請求，則在基于 Request 和回調創建 WebSocket 的時候，OkHttp 會自動發起連接建立的過程。

這也是 WebSocket 與 HTTP 最大的不同。對于 WebSocket，我們可以保存 WebSocket 對象，并在后續多次通過該對象向服務器發送數據。

通過回調可以獲得更多 WebSocket 的狀態變化。在連接建立、收到服務器發送回來的消息、服務器要關閉連接，以及出現 error 時，都能得到通知。不像 HTTP 請求那樣，只在最后得到一個請求成功或者失敗的結果。

后兩點正是 WebSocket 全雙工連接的體現。

OkHttp 的 WebSocket 實現

接著我們來看OkHttp 的 WebSocket 實現。WebSocket 包含兩個部分，分別是握手和數據傳輸，數據傳輸又包括數據的發送，數據的接收，連接的保活，以及連接的關閉等，我們將分別分析這些過程。

連接握手

創建 WebSocket 的過程如下：

public class OkHttpClient implements Cloneable, Call.Factory, WebSocket.Factory {. . . . . .@Override public WebSocket newWebSocket(Request request, WebSocketListener listener) {RealWebSocket webSocket = new RealWebSocket(request, listener, new SecureRandom());webSocket.connect(this);return webSocket;}

在這里會創建一個 RealWebSocket 對象，然后執行其 connect() 方法建立連接。 RealWebSocket 對象的創建過程如下：

public final class RealWebSocket implements WebSocket, WebSocketReader.FrameCallback {. . . . . .public RealWebSocket(Request request, WebSocketListener listener, Random random) {if (!"GET".equals(request.method())) {throw new IllegalArgumentException("Request must be GET: " + request.method());}this.originalRequest = request;this.listener = listener;this.random = random;byte[] nonce = new byte[16];random.nextBytes(nonce);this.key = ByteString.of(nonce).base64();this.writerRunnable = new Runnable() {@Override public void run() {try {while (writeOneFrame()) {}} catch (IOException e) {failWebSocket(e, null);}}};}

這里最主要的是初始化了 key，以備后續連接建立及握手之用。Key 是一個16字節長的隨機數經過 Base64 編碼得到的。此外還初始化了 writerRunnable 等。

連接建立及握手過程如下：

public final class RealWebSocket implements WebSocket, WebSocketReader.FrameCallback {. . . . . .public void connect(OkHttpClient client) {client = client.newBuilder().protocols(ONLY_HTTP1).build();final int pingIntervalMillis = client.pingIntervalMillis();final Request request = originalRequest.newBuilder().header("Upgrade", "websocket").header("Connection", "Upgrade").header("Sec-WebSocket-Key", key).header("Sec-WebSocket-Version", "13").build();call = Internal.instance.newWebSocketCall(client, request);call.enqueue(new Callback() {@Override public void onResponse(Call call, Response response) {try {checkResponse(response);} catch (ProtocolException e) {failWebSocket(e, response);closeQuietly(response);return;}// Promote the HTTP streams into web socket streams.StreamAllocation streamAllocation = Internal.instance.streamAllocation(call);streamAllocation.noNewStreams(); // Prevent connection pooling!Streams streams = streamAllocation.connection().newWebSocketStreams(streamAllocation);// Process all web socket messages.try {listener.onOpen(RealWebSocket.this, response);String name = "OkHttp WebSocket " + request.url().redact();initReaderAndWriter(name, pingIntervalMillis, streams);streamAllocation.connection().socket().setSoTimeout(0);loopReader();} catch (Exception e) {failWebSocket(e, null);}}@Override public void onFailure(Call call, IOException e) {failWebSocket(e, null);}});}

連接建立及握手的過程主要是向服務器發送一個HTTP請求。這個 HTTP 請求的特別之處在于，它包含了如下的一些Headers：

Upgrade: WebSocket Connection: Upgrade Sec-WebSocket-Key: 7wgaspE0Tl7/66o4Dov2kw== Sec-WebSocket-Version: 13

其中 Upgrade 和 Connection header 向服務器表明，請求的目的就是要將客戶端和服務器端的通訊協議從 HTTP 協議升級到 WebSocket 協議，同時在請求處理完成之后，連接不要斷開。Sec-WebSocket-Key header 值正是我們前面看到的key，它是 WebSocket 客戶端發送的一個 base64 編碼的密文，要求服務端必須返回一個對應加密的 “Sec-WebSocket-Accept” 應答，否則客戶端會拋出 “Error during WebSocket handshake” 錯誤，并關閉連接。

來自于 HTTP 服務器的響應到達的時候，即是連接建立大功告成的時候，也就是熱豆腐孰了的時候。

然而，響應到達時，盡管連接已經建立，還要為數據的收發做一些準備。這些準備中的第一步就是檢查 HTTP 響應：

public final class RealWebSocket implements WebSocket, WebSocketReader.FrameCallback {. . . . . .void checkResponse(Response response) throws ProtocolException {if (response.code() != 101) {throw new ProtocolException("Expected HTTP 101 response but was '"+ response.code() + " " + response.message() + "'");}String headerConnection = response.header("Connection");if (!"Upgrade".equalsIgnoreCase(headerConnection)) {throw new ProtocolException("Expected 'Connection' header value 'Upgrade' but was '"+ headerConnection + "'");}String headerUpgrade = response.header("Upgrade");if (!"websocket".equalsIgnoreCase(headerUpgrade)) {throw new ProtocolException("Expected 'Upgrade' header value 'websocket' but was '" + headerUpgrade + "'");}String headerAccept = response.header("Sec-WebSocket-Accept");String acceptExpected = ByteString.encodeUtf8(key + WebSocketProtocol.ACCEPT_MAGIC).sha1().base64();if (!acceptExpected.equals(headerAccept)) {throw new ProtocolException("Expected 'Sec-WebSocket-Accept' header value '"+ acceptExpected + "' but was '" + headerAccept + "'");}}. . . . . .public void failWebSocket(Exception e, 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();}try {listener.onFailure(this, e, response);} finally {closeQuietly(streamsToClose);}}

根據 WebSocket 的協議，服務器端用如下響應，來表示接受建立 WebSocket 連接的請求：

響應碼是 101。

"Connection" header 的值為 "Upgrade"，以表明服務器并沒有在處理完請求之后把連接個斷開。

"Upgrade" header 的值為 "websocket"，以表明服務器接受后面使用 WebSocket 來通信。

"Sec-WebSocket-Accept" header 的值為，key + WebSocketProtocol.ACCEPT_MAGIC 做 SHA1 hash，然后做 base64 編碼，來做服務器接受連接的驗證。關于這部分的設計的詳細信息，可參考 WebSocket 協議規范。

為數據收發做準備的第二步是，初始化用于輸入輸出的 Source 和 Sink。Source 和 Sink 創建于之前發送HTTP請求的時候。這里會阻止在這個連接上再創建新的流。

public final class RealConnection extends Http2Connection.Listener implements Connection {. . . . . .public RealWebSocket.Streams newWebSocketStreams(final StreamAllocation streamAllocation) {return new RealWebSocket.Streams(true, source, sink) {@Override public void close() throws IOException {streamAllocation.streamFinished(true, streamAllocation.codec());}};}

Streams是一個 BufferedSource 和 BufferedSink 的holder：

public final class RealWebSocket implements WebSocket, WebSocketReader.FrameCallback {. . . . . .public abstract static class Streams implements Closeable {public final boolean client;public final BufferedSource source;public final BufferedSink sink;public Streams(boolean client, BufferedSource source, BufferedSink sink) {this.client = client;this.source = source;this.sink = sink;}}

第三步是調用回調 onOpen()。

第四步是初始化 Reader 和 Writer：

public final class RealWebSocket implements WebSocket, WebSocketReader.FrameCallback {. . . . . .public void initReaderAndWriter(String name, long pingIntervalMillis, Streams streams) throws IOException {synchronized (this) {this.streams = streams;this.writer = new WebSocketWriter(streams.client, streams.sink, random);this.executor = new ScheduledThreadPoolExecutor(1, Util.threadFactory(name, false));if (pingIntervalMillis != 0) {executor.scheduleAtFixedRate(new PingRunnable(), pingIntervalMillis, pingIntervalMillis, MILLISECONDS);}if (!messageAndCloseQueue.isEmpty()) {runWriter(); // Send messages that were enqueued before we were connected.}}reader = new WebSocketReader(streams.client, streams.source, this);}

OkHttp使用 WebSocketReader 和 WebSocketWriter 來處理數據的收發。在發送數據時將數據組織成幀，在接收數據時則進行反向擦做，同時處理 WebSocket 的控制消息。

WebSocket 的所有數據發送動作，都會在單線程線程池的線程中，通過 WebSocketWriter 執行。在這里會創建 ScheduledThreadPoolExecutor 用于跑數據的發送操作。WebSocket 協議中主要會傳輸兩種類型的幀，一是控制幀，主要是用于連接保活的 Ping 幀等；二是用戶數據載荷幀。在這里會根據用戶的配置，調度 Ping 幀周期性地發送。我們在調用 WebSocket 的接口發送數據時，數據并不是同步發送的，而是被放在了一個消息隊列中。發送消息的 Runnable 從消息隊列中讀取數據發送。這里會檢查消息隊列中是否有數據，如果有的話，會調度發送消息的 Runnable 執行。

第五步是配置socket的超時時間為0，也就是阻塞IO。

第六步執行 loopReader()。這實際上是進入了消息讀取循環了，也就是數據接收的邏輯了。

數據發送

我們可以通過 WebSocket 接口的 send(String text) 和 send(ByteString bytes) 分別發送文本的和二進制格式的消息。

public final class RealWebSocket implements WebSocket, WebSocketReader.FrameCallback {. . . . . .@Override public boolean send(String text) {if (text == null) throw new NullPointerException("text == null");return send(ByteString.encodeUtf8(text), OPCODE_TEXT);}@Override public boolean send(ByteString bytes) {if (bytes == null) throw new NullPointerException("bytes == null");return send(bytes, OPCODE_BINARY);}private synchronized boolean send(ByteString data, int formatOpcode) {// Don't send new frames after we've failed or enqueued a close frame.if (failed || enqueuedClose) return false;// If this frame overflows the buffer, reject it and close the web socket.if (queueSize + data.size() > MAX_QUEUE_SIZE) {close(CLOSE_CLIENT_GOING_AWAY, null);return false;}// Enqueue the message frame.queueSize += data.size();messageAndCloseQueue.add(new Message(formatOpcode, data));runWriter();return true;}. . . . . .private void runWriter() {assert (Thread.holdsLock(this));if (executor != null) {executor.execute(writerRunnable);}}

可以看到我們調用發送數據的接口時，做的事情主要是將數據格式化，構造消息，放進一個消息隊列，然后調度 writerRunnable 執行。

此外，值得注意的是，當消息隊列中的未發送數據超出最大大小限制，WebSocket 連接會被直接關閉。對于發送失敗過或被關閉了的 WebSocket，將無法再發送信息。

在 writerRunnable 中會循環調用 writeOneFrame() 逐幀發送數據，直到數據發完，或發送失敗。在 WebSocket 協議中，客戶端需要發送 四種類型 的幀：

PING 幀

PONG 幀

CLOSE 幀

MESSAGE 幀

PING幀用于連接保活，它的發送是在 PingRunnable 中執行的，在初始化 Reader 和 Writer 的時候，就會根據設置調度執行或不執行。除PING 幀外的其它 三種 幀，都在 writeOneFrame() 中發送。PONG 幀是對服務器發過來的 PING 幀的響應，同樣用于保活連接。后面我們在分析連接的保活時會更詳細的分析 PING 和 PONG 這兩種幀。CLOSE 幀用于關閉連接，稍后我們在分析連接關閉過程時再來詳細地分析。

這里我們主要關注用戶數據發送的部分。PONG 幀具有最高的發送優先級。在沒有PONG 幀需要發送時，writeOneFrame() 從消息隊列中取出一條消息，如果消息不是 CLOSE 幀，則主要通過如下的過程進行發送：

public final class RealWebSocket implements WebSocket, WebSocketReader.FrameCallback {. . . . . .boolean writeOneFrame() throws IOException {WebSocketWriter writer;ByteString pong;Object messageOrClose = null;int receivedCloseCode = -1;String receivedCloseReason = null;Streams streamsToClose = null;synchronized (RealWebSocket.this) {if (failed) {return false; // Failed web socket.}writer = this.writer;pong = pongQueue.poll();if (pong == null) {messageOrClose = messageAndCloseQueue.poll();. . . . . .} else if (messageOrClose instanceof Message) {ByteString data = ((Message) messageOrClose).data;BufferedSink sink = Okio.buffer(writer.newMessageSink(((Message) messageOrClose).formatOpcode, data.size()));sink.write(data);sink.close();synchronized (this) {queueSize -= data.size();}} else if (messageOrClose instanceof Close) {

數據發送的過程可以總結如下：

創建一個 BufferedSink 用于數據發送。

將數據寫入前面創建的 BufferedSink 中。

關閉 BufferedSink。

更新 queueSize 以正確地指示未發送數據的長度。

這里面的玄機主要在創建的 BufferedSink。創建的 Sink 是一個 FrameSink：

static void toggleMask(byte[] buffer, long byteCount, byte[] key, long frameBytesRead) {int keyLength = key.length;for (int i = 0; i < byteCount; i++, frameBytesRead++) {int keyIndex = (int) (frameBytesRead % keyLength);buffer[i] = (byte) (buffer[i] ^ key[keyIndex]);}}. . . . . .Sink newMessageSink(int formatOpcode, long contentLength) {if (activeWriter) {throw new IllegalStateException("Another message writer is active. Did you call close()?");}activeWriter = true;// Reset FrameSink state for a new writer.frameSink.formatOpcode = formatOpcode;frameSink.contentLength = contentLength;frameSink.isFirstFrame = true;frameSink.closed = false;return frameSink;}void writeMessageFrameSynchronized(int formatOpcode, long byteCount, boolean isFirstFrame,boolean isFinal) throws IOException {assert Thread.holdsLock(this);if (writerClosed) throw new IOException("closed");int b0 = isFirstFrame ? formatOpcode : OPCODE_CONTINUATION;if (isFinal) {b0 |= B0_FLAG_FIN;}sink.writeByte(b0);int b1 = 0;if (isClient) {b1 |= B1_FLAG_MASK;}if (byteCount <= PAYLOAD_BYTE_MAX) {b1 |= (int) byteCount;sink.writeByte(b1);} else if (byteCount <= PAYLOAD_SHORT_MAX) {b1 |= PAYLOAD_SHORT;sink.writeByte(b1);sink.writeShort((int) byteCount);} else {b1 |= PAYLOAD_LONG;sink.writeByte(b1);sink.writeLong(byteCount);}if (isClient) {random.nextBytes(maskKey);sink.write(maskKey);for (long written = 0; written < byteCount; ) {int toRead = (int) Math.min(byteCount, maskBuffer.length);int read = buffer.read(maskBuffer, 0, toRead);if (read == -1) throw new AssertionError();toggleMask(maskBuffer, read, maskKey, written);sink.write(maskBuffer, 0, read);written += read;}} else {sink.write(buffer, byteCount);}sink.emit();}final class FrameSink implements Sink {int formatOpcode;long contentLength;boolean isFirstFrame;boolean closed;@Override public void write(Buffer source, long byteCount) throws IOException {if (closed) throw new IOException("closed");buffer.write(source, byteCount);// Determine if this is a buffered write which we can defer until close() flushes.boolean deferWrite = isFirstFrame&& contentLength != -1&& buffer.size() > contentLength - 8192 /* segment size */;long emitCount = buffer.completeSegmentByteCount();if (emitCount > 0 && !deferWrite) {synchronized (WebSocketWriter.this) {writeMessageFrameSynchronized(formatOpcode, emitCount, isFirstFrame, false /* final */);}isFirstFrame = false;}}@Override public void flush() throws IOException {if (closed) throw new IOException("closed");synchronized (WebSocketWriter.this) {writeMessageFrameSynchronized(formatOpcode, buffer.size(), isFirstFrame, false /* final */);}isFirstFrame = false;}@Override public Timeout timeout() {return sink.timeout();}@SuppressWarnings("PointlessBitwiseExpression")@Override public void close() throws IOException {if (closed) throw new IOException("closed");synchronized (WebSocketWriter.this) {writeMessageFrameSynchronized(formatOpcode, buffer.size(), isFirstFrame, true /* final */);}closed = true;activeWriter = false;}}

FrameSink 的 write() 會先將數據寫如一個 Buffer 中，然后再從這個 Buffer 中讀取數據來發送。如果是第一次發送數據，同時剩余要發送的數據小于 8192 字節時，會延遲執行實際的數據發送，等 close() 時刷新。根據 RealWebSocket 的 writeOneFrame() 的邏輯，在 write() 時，總是寫入整個消息的所有數據，因而，在 FrameSink 的 write() 中總是不會發送數據的。

writeMessageFrameSynchronized() 將用戶數據格式化并發送出去。規范中定義的數據格式如下：

0 1 2 30 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1+-+-+-+-+-------+-+-------------+-------------------------------+|F|R|R|R| opcode|M| Payload len | Extended payload length ||I|S|S|S| (4) |A| (7) | (16/64) ||N|V|V|V| |S| | (if payload len==126/127) || |1|2|3| |K| | |+-+-+-+-+-------+-+-------------+ - - - - - - - - - - - - - - - +| Extended payload length continued, if payload len == 127 |+ - - - - - - - - - - - - - - - +-------------------------------+| |Masking-key, if MASK set to 1 |+-------------------------------+-------------------------------+| Masking-key (continued) | Payload Data |+-------------------------------- - - - - - - - - - - - - - - - +: Payload Data continued ... :+ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - +| Payload Data continued ... |+---------------------------------------------------------------+

基本結構為：

第一個字節是 meta data 控制位，包括四位的操作碼，用于指明這是否是消息的最后一幀的FIN位及三個保留位。

第二個字節包括掩碼位，和載荷長度或載荷長度指示。只有載荷長度比較小，在 127 以內時，載荷長度才會包含在這個字節中。否則這個字節中將包含載荷長度指示的位。

可選的載荷長度。載荷長度大于127時，幀中會專門有一些字節來描述載荷的長度。載荷長度具體占用幾個自己，因載荷的實際長度而異。

可選的掩碼字節。客戶端發送的幀，設置掩碼指示位，并包含四個字節的掩碼字節。

載荷數據。客戶端發送的數據，會將原始的數據與掩碼字節做異或之后再發送。

關于幀格式的更詳細信息，可以參考 WebSocket Protocol 規范。

數據的接收

如我們前面看到的， 在握手的HTTP請求返回之后，會在HTTP請求的回調里，啟動消息讀取循環 loopReader()：

public final class RealWebSocket implements WebSocket, WebSocketReader.FrameCallback {. . . . . ./** Receive frames until there are no more. Invoked only by the reader thread. */public void loopReader() throws IOException {while (receivedCloseCode == -1) {// This method call results in one or more onRead* methods being called on this thread.reader.processNextFrame();}}

在這個循環中，不斷通過 WebSocketReader 的 processNextFrame() 讀取消息，直到收到了關閉連接的消息。

final class WebSocketReader {public interface FrameCallback {void onReadMessage(String text) throws IOException;void onReadMessage(ByteString bytes) throws IOException;void onReadPing(ByteString buffer);void onReadPong(ByteString buffer);void onReadClose(int code, String reason);}. . . . . .void processNextFrame() throws IOException {readHeader();if (isControlFrame) {readControlFrame();} else {readMessageFrame();}}private void readHeader() throws IOException {if (closed) throw new IOException("closed");// Disable the timeout to read the first byte of a new frame.int b0;long timeoutBefore = source.timeout().timeoutNanos();source.timeout().clearTimeout();try {b0 = source.readByte() & 0xff;} finally {source.timeout().timeout(timeoutBefore, TimeUnit.NANOSECONDS);}opcode = b0 & B0_MASK_OPCODE;isFinalFrame = (b0 & B0_FLAG_FIN) != 0;isControlFrame = (b0 & OPCODE_FLAG_CONTROL) != 0;// Control frames must be final frames (cannot contain continuations).if (isControlFrame && !isFinalFrame) {throw new ProtocolException("Control frames must be final.");}boolean reservedFlag1 = (b0 & B0_FLAG_RSV1) != 0;boolean reservedFlag2 = (b0 & B0_FLAG_RSV2) != 0;boolean reservedFlag3 = (b0 & B0_FLAG_RSV3) != 0;if (reservedFlag1 || reservedFlag2 || reservedFlag3) {// Reserved flags are for extensions which we currently do not support.throw new ProtocolException("Reserved flags are unsupported.");}int b1 = source.readByte() & 0xff;isMasked = (b1 & B1_FLAG_MASK) != 0;if (isMasked == isClient) {// Masked payloads must be read on the server. Unmasked payloads must be read on the client.throw new ProtocolException(isClient? "Server-sent frames must not be masked.": "Client-sent frames must be masked.");}// Get frame length, optionally reading from follow-up bytes if indicated by special values.frameLength = b1 & B1_MASK_LENGTH;if (frameLength == PAYLOAD_SHORT) {frameLength = source.readShort() & 0xffffL; // Value is unsigned.} else if (frameLength == PAYLOAD_LONG) {frameLength = source.readLong();if (frameLength < 0) {throw new ProtocolException("Frame length 0x" + Long.toHexString(frameLength) + " > 0x7FFFFFFFFFFFFFFF");}}frameBytesRead = 0;if (isControlFrame && frameLength > PAYLOAD_BYTE_MAX) {throw new ProtocolException("Control frame must be less than " + PAYLOAD_BYTE_MAX + "B.");}if (isMasked) {// Read the masking key as bytes so that they can be used directly for unmasking.source.readFully(maskKey);}}

processNextFrame() 先讀取 Header 的兩個字節，然后根據 Header 的信息，讀取數據內容。

在讀取 Header 時，讀的第一個字節是同步的不計超時時間的。WebSocketReader 從 Header 中，獲取到這個幀是不是消息的最后一幀，消息的類型，是否有掩碼字節，保留位，幀的長度，以及掩碼字節等信息。WebSocket 通過掩碼位和掩碼字節來區分數據是從客戶端發送給服務器的，還是服務器發送給客戶端的。這里會根據協議，對這些信息進行有效性一致性檢驗，若不一致則會拋出 ProtocolException。

WebSocketReader 同步讀取時的調用棧如下：

Reader Thread

通過幀的 Header 確定了是數據幀，則會執行 readMessageFrame() 讀取消息幀：

final class WebSocketReader {. . . . . .private void readMessageFrame() throws IOException {int opcode = this.opcode;if (opcode != OPCODE_TEXT && opcode != OPCODE_BINARY) {throw new ProtocolException("Unknown opcode: " + toHexString(opcode));}Buffer message = new Buffer();readMessage(message);if (opcode == OPCODE_TEXT) {frameCallback.onReadMessage(message.readUtf8());} else {frameCallback.onReadMessage(message.readByteString());}}/** Read headers and process any control frames until we reach a non-control frame. */void readUntilNonControlFrame() throws IOException {while (!closed) {readHeader();if (!isControlFrame) {break;}readControlFrame();}}/*** Reads a message body into across one or more frames. Control frames that occur between* fragments will be processed. If the message payload is masked this will unmask as it's being* processed.*/private void readMessage(Buffer sink) throws IOException {while (true) {if (closed) throw new IOException("closed");if (frameBytesRead == frameLength) {if (isFinalFrame) return; // We are exhausted and have no continuations.readUntilNonControlFrame();if (opcode != OPCODE_CONTINUATION) {throw new ProtocolException("Expected continuation opcode. Got: " + toHexString(opcode));}if (isFinalFrame && frameLength == 0) {return; // Fast-path for empty final frame.}}long toRead = frameLength - frameBytesRead;long read;if (isMasked) {toRead = Math.min(toRead, maskBuffer.length);read = source.read(maskBuffer, 0, (int) toRead);if (read == -1) throw new EOFException();toggleMask(maskBuffer, read, maskKey, frameBytesRead);sink.write(maskBuffer, 0, (int) read);} else {read = source.read(sink, toRead);if (read == -1) throw new EOFException();}frameBytesRead += read;}}

這個過程中，會讀取一條消息包含的所有數據幀。按照 WebSocket 的標準，包含用戶數據的消息數據幀可以和控制幀交替發送；但消息之間的數據幀不可以。因而在這個過程中，若遇到了控制幀，則會先讀取控制幀進行處理，然后繼續讀取消息的數據幀，直到讀取了消息的所有數據幀。

掩碼位和掩碼字節，對于客戶端而言，發送的數據中包含這些東西，在接收的數據中不包含這些；對于服務器而言，則是在接收的數據中包含這些，發送的數據中不包含。OkHttp 既支持服務器開發，也支持客戶端開發，因而可以看到對于掩碼位和掩碼字節完整的處理。

在一個消息讀取完成之后，會通過回調 FrameCallback 將讀取的內容通知出去。

final class WebSocketReader {. . . . . .WebSocketReader(boolean isClient, BufferedSource source, FrameCallback frameCallback) {if (source == null) throw new NullPointerException("source == null");if (frameCallback == null) throw new NullPointerException("frameCallback == null");this.isClient = isClient;this.source = source;this.frameCallback = frameCallback;}

這一事件會通知到 RealWebSocket。

public final class RealWebSocket implements WebSocket, WebSocketReader.FrameCallback {. . . . . .@Override public void onReadMessage(String text) throws IOException {listener.onMessage(this, text);}@Override public void onReadMessage(ByteString bytes) throws IOException {listener.onMessage(this, bytes);}

在 RealWebSocket 中，這一事件又被通知到我們在應用程序中創建的回調 WebSocketListener。

連接的保活

連接的保活通過 PING 幀和 PONG 幀來實現。如我們前面看到的，若用戶設置了 PING 幀的發送周期，在握手的HTTP請求返回時，消息讀取循環開始前會調度 PingRunnable 周期性的向服務器發送 PING 幀：

public final class RealWebSocket implements WebSocket, WebSocketReader.FrameCallback {. . . . . .private final class PingRunnable implements Runnable {PingRunnable() {}@Override public void run() {writePingFrame();}}void writePingFrame() {WebSocketWriter writer;synchronized (this) {if (failed) return;writer = this.writer;}try {writer.writePing(ByteString.EMPTY);} catch (IOException e) {failWebSocket(e, null);}}

在 PingRunnable 中，通過 WebSocketWriter 發送 PING 幀：

final class WebSocketWriter {. . . . . ./** Send a ping with the supplied {@code payload}. */void writePing(ByteString payload) throws IOException {synchronized (this) {writeControlFrameSynchronized(OPCODE_CONTROL_PING, payload);}}. . . . . .private void writeControlFrameSynchronized(int opcode, ByteString payload) throws IOException {assert Thread.holdsLock(this);if (writerClosed) throw new IOException("closed");int length = payload.size();if (length > PAYLOAD_BYTE_MAX) {throw new IllegalArgumentException("Payload size must be less than or equal to " + PAYLOAD_BYTE_MAX);}int b0 = B0_FLAG_FIN | opcode;sink.writeByte(b0);int b1 = length;if (isClient) {b1 |= B1_FLAG_MASK;sink.writeByte(b1);random.nextBytes(maskKey);sink.write(maskKey);byte[] bytes = payload.toByteArray();toggleMask(bytes, bytes.length, maskKey, 0);sink.write(bytes);} else {sink.writeByte(b1);sink.write(payload);}sink.flush();}

PING 幀是一個不包含載荷的控制幀。關于掩碼位和掩碼字節的設置，與消息的數據幀相同。即客戶端發送的幀，設置掩碼位，幀中包含掩碼字節；服務器發送的幀，不設置掩碼位，幀中不包含掩碼字節。

通過 WebSocket 通信的雙方，在收到對方發來的 PING 幀時，需要用PONG幀來回復。在 WebSocketReader 的 readControlFrame() 中可以看到這一點：

final class WebSocketReader {. . . . . .private void readControlFrame() throws IOException {Buffer buffer = new Buffer();if (frameBytesRead < frameLength) {if (isClient) {source.readFully(buffer, frameLength);} else {while (frameBytesRead < frameLength) {int toRead = (int) Math.min(frameLength - frameBytesRead, maskBuffer.length);int read = source.read(maskBuffer, 0, toRead);if (read == -1) throw new EOFException();toggleMask(maskBuffer, read, maskKey, frameBytesRead);buffer.write(maskBuffer, 0, read);frameBytesRead += read;}}}switch (opcode) {case OPCODE_CONTROL_PING:frameCallback.onReadPing(buffer.readByteString());break;case OPCODE_CONTROL_PONG:frameCallback.onReadPong(buffer.readByteString());break;

PING 幀和 PONG 幀都不帶載荷，控制幀讀寫時對于載荷長度的處理，都是為 CLOSE 幀做的。因而針對 PING 幀和 PONG 幀，除了 Header 外， readControlFrame() 實際上無需再讀取任何數據，但它會將這些事件通知出去：

public final class RealWebSocket implements WebSocket, WebSocketReader.FrameCallback {. . . . . .@Override public synchronized void onReadPing(ByteString payload) {// Don't respond to pings after we've failed or sent the close frame.if (failed || (enqueuedClose && messageAndCloseQueue.isEmpty())) return;pongQueue.add(payload);runWriter();pingCount++;}@Override public synchronized void onReadPong(ByteString buffer) {// This API doesn't expose pings.pongCount++;}

可見在收到 PING 幀的時候，總是會發一個 PONG 幀出去，且通常其沒有載荷數據。在收到一個 PONG 幀時，則通常只是記錄一下，然后什么也不做。如我們前面所見，PONG 幀在 writerRunnable 中被發送出去：

public final class RealWebSocket implements WebSocket, WebSocketReader.FrameCallback {. . . . . .if (pong != null) {writer.writePong(pong);} else if (messageOrClose instanceof Message) {

PONG 幀的發送與 PING 幀的非常相似：

final class WebSocketWriter {. . . . . ./** Send a pong with the supplied {@code payload}. */void writePong(ByteString payload) throws IOException {synchronized (this) {writeControlFrameSynchronized(OPCODE_CONTROL_PONG, payload);}}

連接的關閉

連接的關閉，與數據發送的過程頗有幾分相似之處。通過 WebSocket 接口的 close(int code, String reason) 我們可以關閉一個 WebSocket 連接：

public final class RealWebSocket implements WebSocket, WebSocketReader.FrameCallback {. . . . . .@Override public boolean close(int code, String reason) {return close(code, reason, CANCEL_AFTER_CLOSE_MILLIS);}synchronized boolean close(int code, String reason, long cancelAfterCloseMillis) {validateCloseCode(code);ByteString reasonBytes = null;if (reason != null) {reasonBytes = ByteString.encodeUtf8(reason);if (reasonBytes.size() > CLOSE_MESSAGE_MAX) {throw new IllegalArgumentException("reason.size() > " + CLOSE_MESSAGE_MAX + ": " + reason);}}if (failed || enqueuedClose) return false;// Immediately prevent further frames from being enqueued.enqueuedClose = true;// Enqueue the close frame.messageAndCloseQueue.add(new Close(code, reasonBytes, cancelAfterCloseMillis));runWriter();return true;}

在執行關閉連接動作前，會先檢查一下 close code 的有效性在合法范圍內。關于不同 close code 的詳細說明，可以參考 WebSocket 協議規范。

檢查完了之后，會構造一個 Close 消息放入發送消息隊列，并調度 writerRunnable 執行。Close 消息可以帶有不超出 123 字節的字符串，以作為 Close message，來說明連接關閉的原因。

連接的關閉分為主動關閉和被動關閉。客戶端先向服務器發送一個 CLOSE 幀，然后服務器恢復一個 CLOSE 幀，對于客戶端而言，這個過程為主動關閉；反之則為對客戶端而言則為被動關閉。

在 writerRunnable 執行的 writeOneFrame() 實際發送 CLOSE 幀：

public final class RealWebSocket implements WebSocket, WebSocketReader.FrameCallback {. . . . . .messageOrClose = messageAndCloseQueue.poll();if (messageOrClose instanceof Close) {receivedCloseCode = this.receivedCloseCode;receivedCloseReason = this.receivedCloseReason;if (receivedCloseCode != -1) {streamsToClose = this.streams;this.streams = null;this.executor.shutdown();} else {// When we request a graceful close also schedule a cancel of the websocket.cancelFuture = executor.schedule(new CancelRunnable(),((Close) messageOrClose).cancelAfterCloseMillis, MILLISECONDS);}} else if (messageOrClose == null) {return false; // The queue is exhausted.}}. . . . . .} else if (messageOrClose instanceof Close) {Close close = (Close) messageOrClose;writer.writeClose(close.code, close.reason);// We closed the writer: now both reader and writer are closed.if (streamsToClose != null) {listener.onClosed(this, receivedCloseCode, receivedCloseReason);}} else {

發送 CLOSE 幀也分為主動關閉的發送還是被動關閉的發送。
對于被動關閉，在發送完 CLOSE 幀之后，連接被最終關閉，因而，發送 CLOSE 幀之前，這里會停掉發送消息用的 executor。而在發送之后，則會通過 onClosed() 通知用戶。

而對于主動關閉，則在發送前會調度 CancelRunnable 的執行，發送后不會通過 onClosed() 通知用戶。

final class WebSocketWriter {. . . . . .void writeClose(int code, ByteString reason) throws IOException {ByteString payload = ByteString.EMPTY;if (code != 0 || reason != null) {if (code != 0) {validateCloseCode(code);}Buffer buffer = new Buffer();buffer.writeShort(code);if (reason != null) {buffer.write(reason);}payload = buffer.readByteString();}synchronized (this) {try {writeControlFrameSynchronized(OPCODE_CONTROL_CLOSE, payload);} finally {writerClosed = true;}}}

將 CLOSE 幀發送到網絡的過程與 PING 和 PONG 幀的頗為相似，僅有的差別就是 CLOSE 幀有載荷。關于掩碼位和掩碼自己的規則，同樣適用于 CLOSE 幀的發送。

CLOSE 的讀取在 WebSocketReader 的 readControlFrame()中：

final class WebSocketReader {. . . . . .private void readControlFrame() throws IOException {Buffer buffer = new Buffer();if (frameBytesRead < frameLength) {if (isClient) {source.readFully(buffer, frameLength);} else {while (frameBytesRead < frameLength) {int toRead = (int) Math.min(frameLength - frameBytesRead, maskBuffer.length);int read = source.read(maskBuffer, 0, toRead);if (read == -1) throw new EOFException();toggleMask(maskBuffer, read, maskKey, frameBytesRead);buffer.write(maskBuffer, 0, read);frameBytesRead += read;}}}switch (opcode) {. . . . . .case OPCODE_CONTROL_CLOSE:int code = CLOSE_NO_STATUS_CODE;String reason = "";long bufferSize = buffer.size();if (bufferSize == 1) {throw new ProtocolException("Malformed close payload length of 1.");} else if (bufferSize != 0) {code = buffer.readShort();reason = buffer.readUtf8();String codeExceptionMessage = WebSocketProtocol.closeCodeExceptionMessage(code);if (codeExceptionMessage != null) throw new ProtocolException(codeExceptionMessage);}frameCallback.onReadClose(code, reason);closed = true;break;default:throw new ProtocolException("Unknown control opcode: " + toHexString(opcode));}}

讀到 CLOSE 幀時，WebSocketReader 會將這一事件通知出去：

public final class RealWebSocket implements WebSocket, WebSocketReader.FrameCallback {. . . . . .@Override public void onReadClose(int code, String reason) {if (code == -1) throw new IllegalArgumentException();Streams toClose = null;synchronized (this) {if (receivedCloseCode != -1) throw new IllegalStateException("already closed");receivedCloseCode = code;receivedCloseReason = reason;if (enqueuedClose && messageAndCloseQueue.isEmpty()) {toClose = this.streams;this.streams = null;if (cancelFuture != null) cancelFuture.cancel(false);this.executor.shutdown();}}try {listener.onClosing(this, code, reason);if (toClose != null) {listener.onClosed(this, code, reason);}} finally {closeQuietly(toClose);}}

對于收到的 CLOSE 幀處理同樣分為主動關閉的情況和被動關閉的情況。與 CLOSE 發送時的情形正好相反，若是主動關閉，則在收到 CLOSE 幀之后，WebSocket 連接最終斷開，因而需要停掉executor，被動關閉則暫時不需要。

收到 CLOSE 幀，總是會通過 onClosing() 將事件通知出去。

對于主動關閉的情形，最后還會通過 onClosed() 通知用戶，連接已經最終關閉。

關于 WebSocket 的 CLOSE 幀的更多說明，可以參考 WebSocket協議規范。

WebSocket連接的生命周期

總結一下 WebSocket 連接的生命周期：

連接通過一個HTTP請求握手并建立連接。WebSocket 連接可以理解為是通過HTTP請求建立的普通TCP連接。

WebSocket 做了二進制分幀。WebSocket 連接中收發的數據以幀為單位。主要有用于連接保活的控制幀 PING 和 PONG，用于用戶數據發送的 MESSAGE 幀，和用于關閉連接的控制幀 CLOSE。

連接建立之后，通過 PING 幀和 PONG 幀做連接保活。

一次 send 數據，被封為一個消息，通過一個或多個 MESSAGE幀進行發送。一個消息的幀和控制幀可以交叉發送，不同消息的幀之間不可以。

WebSocket 連接的兩端相互發送一個 CLOSE 幀以最終關閉連接。

關于 WebSocket 的詳細信息，可以參考 WebSocket協議規范。

參考文檔

WebSocket 協議規范
WebSocket 實戰
使用 HTML5 WebSocket 構建實時 Web 應用
WebSocket Client Example with OkHttp

總結

以上是生活随笔為你收集整理的OkHttp实现分析之Websocket的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。