系统学习详见OKhttp源码解析详解系列
连接的保活通过 PING 帧和 PONG 帧来实现。
1 周期性的向服务器发送 PING 帧
okhttp之旅（十二）--websocket的使用及建立连接
中2.4.4 第四步是初始化 Reader 和 Writer时开启线程周期性发送 PING 帧
public final class RealWebSocket implements WebSocket, WebSocketReader.FrameCallback {
public void initReaderAndWriter(String name, 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);
}
}
2 PingRunnable 中，通过WebSocketWriter 发送 PING 帧：
- 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;
int failedPing;
synchronized (this) {
if (failed) return;
writer = this.writer;
failedPing = awaitingPong ? sentPingCount : -1;
sentPingCount++;
awaitingPong = true;
}
if (failedPing != -1) {
failWebSocket(new SocketTimeoutException("sent ping but didn't receive pong within "
+ pingIntervalMillis + "ms (after " + (failedPing - 1) + " successful ping/pongs)"),
null);
return;
}
try {
writer.writePing(ByteString.EMPTY);
} catch (IOException e) {
failWebSocket(e, null);
}
}
}
final class WebSocketWriter {
/** Send a ping with the supplied {@code payload}. */
void writePing(ByteString payload) throws IOException {
writeControlFrame(OPCODE_CONTROL_PING, payload);
}
private void writeControlFrame(int opcode, ByteString payload) throws IOException {
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;
sinkBuffer.writeByte(b0);
int b1 = length;
if (isClient) {
b1 |= B1_FLAG_MASK;
sinkBuffer.writeByte(b1);
random.nextBytes(maskKey);
sinkBuffer.write(maskKey);
if (length > 0) {
long payloadStart = sinkBuffer.size();
sinkBuffer.write(payload);
sinkBuffer.readAndWriteUnsafe(maskCursor);
maskCursor.seek(payloadStart);
toggleMask(maskCursor, maskKey);
maskCursor.close();
}
} else {
sinkBuffer.writeByte(b1);
sinkBuffer.write(payload);
}
sink.flush();
}
}
3 收到对方发来的 PING 帧时，需要用PONG帧来回复
- 通过 WebSocket 通信的双方，在收到对方发来的 PING 帧时，需要用PONG帧来回复。在 WebSocketReader 的 readControlFrame() 中可以看到这一点：
- PING 帧和 PONG 帧都不带载荷，控制帧读写时对于载荷长度的处理，都是为 CLOSE 帧做的。
- 因而针对 PING 帧和 PONG 帧，除了 Header 外， readControlFrame() 实际上无需再读取任何数据，但它会将这些事件通知出去：
final class WebSocketReader {
private void readControlFrame() throws IOException {
if (frameLength > 0) {
source.readFully(controlFrameBuffer, frameLength);
if (!isClient) {
controlFrameBuffer.readAndWriteUnsafe(maskCursor);
maskCursor.seek(0);
toggleMask(maskCursor, maskKey);
maskCursor.close();
}
}
switch (opcode) {
case OPCODE_CONTROL_PING:
frameCallback.onReadPing(controlFrameBuffer.readByteString());
break;
case OPCODE_CONTROL_PONG:
frameCallback.onReadPong(controlFrameBuffer.readByteString());
break;
case OPCODE_CONTROL_CLOSE:
int code = CLOSE_NO_STATUS_CODE;
String reason = "";
long bufferSize = controlFrameBuffer.size();
if (bufferSize == 1) {
throw new ProtocolException("Malformed close payload length of 1.");
} else if (bufferSize != 0) {
code = controlFrameBuffer.readShort();
reason = controlFrameBuffer.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));
}
}
}
- 在收到 PING 帧的时候，总是会发一个 PONG 帧出去，且通常其没有载荷数据。在收到一个 PONG 帧时，则通常只是记录一下，然后什么也不做。
- PONG 帧在 writerRunnable 中被发送出去：
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();
receivedPingCount++;
}
@Override
public synchronized void onReadPong(ByteString buffer) {
// This API doesn't expose pings.
receivedPongCount++;
awaitingPong = false;
}
private void runWriter() {
assert (Thread.holdsLock(this));
if (executor != null) {
executor.execute(writerRunnable);
}
}
}
- PONG 帧在 writerRunnable 中被发送出去：
public final class RealWebSocket implements WebSocket, WebSocketReader.FrameCallback {
public RealWebSocket(Request request, WebSocketListener listener, Random random,
long pingIntervalMillis) {
...
//初始化了 writerRunnable
this.writerRunnable = new Runnable() {
@Override
public void run() {
try {
while (writeOneFrame()) {
}
} catch (IOException e) {
failWebSocket(e, null);
}
}
};
}
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; // websocket连接失败，跳出循环
}
writer = this.writer;
pong = pongQueue.poll();
...
}
try {
if (pong != null) {
writer.writePong(pong);
} else if (messageOrClose instanceof Message) {
ByteString data = ((Message) messageOrClose).data;
//这里将数据转化为可供websocket交互的格式
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) {
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 {
throw new AssertionError();
}
return true;
} finally {
//释放资源
closeQuietly(streamsToClose);
}
}
}
final class WebSocketWriter {
/** Send a pong with the supplied {@code payload}. */
void writePong(ByteString payload) throws IOException {
writeControlFrame(OPCODE_CONTROL_PONG, payload);
}
private void writeControlFrame(int opcode, ByteString payload) throws IOException {
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;
sinkBuffer.writeByte(b0);
int b1 = length;
if (isClient) {
b1 |= B1_FLAG_MASK;
sinkBuffer.writeByte(b1);
random.nextBytes(maskKey);
sinkBuffer.write(maskKey);
if (length > 0) {
long payloadStart = sinkBuffer.size();
sinkBuffer.write(payload);
sinkBuffer.readAndWriteUnsafe(maskCursor);
maskCursor.seek(payloadStart);
toggleMask(maskCursor, maskKey);
maskCursor.close();
}
} else {
sinkBuffer.writeByte(b1);
sinkBuffer.write(payload);
}
sink.flush();
}
}
