1. websocket

• 1.客户端与服务端建连接
• 2.客户端向服务端发送消息
• 3.服务端在接受消息后以“response-接受的消息内容“的形式返回给客户端
• 4.当服务端收到第5条信息的时候，主动关闭与客户端的连接

客户端代码

​private void clientWebSocket(String url) {OkHttpClient client = new OkHttpClient.Builder().build();//构造request对象Request request = new Request.Builder().url(url).build();//建立连接client.newWebSocket(request, new WebSocketListener() {//当远程对等方接受Web套接字并可能开始传输消息时调用。@Overridepublic void onOpen(WebSocket webSocket, Response response) {super.onOpen(webSocket, response);webSocket.send("发送消息");}//当收到文本（类型{@code 0x1}）消息时调用@Overridepublic void onMessage(WebSocket webSocket, String text) {super.onMessage(webSocket, text);}//当收到二进制（类型为{@code 0x2}）消息时调用。@Overridepublic void onMessage(WebSocket webSocket, ByteString bytes) {super.onMessage(webSocket, bytes);}//当远程对等体指示不再有传入的消息将被传输时调用。@Overridepublic void onClosing(WebSocket webSocket, int code, String reason) {super.onClosing(webSocket, code, reason);}//当两个对等方都表示不再传输消息并且连接已成功释放时调用。 没有进一步的电话给这位听众。@Overridepublic void onClosed(WebSocket webSocket, int code, String reason) {super.onClosed(webSocket, code, reason);}//由于从网络读取或向网络写入错误而关闭Web套接字时调用。// 传出和传入的消息都可能丢失。 没有进一步的电话给这位听众。@Overridepublic void onFailure(WebSocket webSocket, Throwable t, @Nullable Response response) {super.onFailure(webSocket, t, response);}});}​

2 客户端与服务端建连接

2.1创建一个 RealWebSocket 对象，然后执行其 connect() 方法建立连接。

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

2.2 真真的RealWebSocket

主要的是初始化了 key，以备后续连接建立及握手之用。Key 是一个16字节长的随机数经过 Base64 编码得到的。此外还初始化了 writerRunnable 等。

​
public final class RealWebSocket implements WebSocket, WebSocketReader.FrameCallbackpublic RealWebSocket(Request request, WebSocketListener listener, Random random,long pingIntervalMillis) {//使用get的方式进行握手if (!"GET".equals(request.method())) {throw new IllegalArgumentException("Request must be GET: " + request.method());}this.originalRequest = request;this.listener = listener;this.random = random;this.pingIntervalMillis = pingIntervalMillis;//初始化key，以备后续连接建立及握手之用。byte[] nonce = new byte[16];random.nextBytes(nonce);this.key = ByteString.of(nonce).base64();//初始化了 writerRunnablethis.writerRunnable = new Runnable() {@Overridepublic void run() {try {while (writeOneFrame()) {}} catch (IOException e) {failWebSocket(e, null);}}};}
}​

2.3 握手建立连接

• 1.创建一个 RealWebSocket 对象，然后执行其 connect() 方法建立连接。
• 2.连接建立及握手的过程主要是向服务器发送一个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” 错误，并关闭连接。
• 3.来自于 HTTP 服务器的响应到达的时候，即是连接建立大功告成的时候

​
public final class RealWebSocket implements WebSocket, WebSocketReader.FrameCallback {public void connect(OkHttpClient client) {client = client.newBuilder().eventListener(EventListener.NONE).protocols(ONLY_HTTP1).build();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);/*** WebSocket协议首先会通过发送一个http请求来完成一个握手的过程* 客户端发送一个请求协议升级的get请求给服务端* 服务端如果支持的话会返回http code 为101，表示可以切换到对应的协议*/call.enqueue(new Callback() {@Overridepublic void onResponse(Call call, Response response) {try {//第一步就是检查 HTTP 响应checkResponse(response);} catch (ProtocolException e) {failWebSocket(e, response);closeQuietly(response);return;}//将HTTP流推广到Web套接字流中StreamAllocation streamAllocation = Internal.instance.streamAllocation(call);streamAllocation.noNewStreams(); //阻止连接池//第二步：初始化用于输入输出的 Source 和 Sink。// Source 和 Sink 创建于之前发送HTTP请求的时候。这里会阻止在这个连接上再创建新的流。Streams streams = streamAllocation.connection().newWebSocketStreams(streamAllocation);// Process all web socket messages.try {//第三步是调用回调 onOpen()。listener.onOpen(RealWebSocket.this, response);//第四步是初始化 Reader 和 Writer：String name = "OkHttp WebSocket " + request.url().redact();initReaderAndWriter(name, streams);//第五步是配置socket的超时时间为0，也就是阻塞IO。streamAllocation.connection().socket().setSoTimeout(0);//第六步执行 loopReader()。这实际上是进入了消息读取循环了，也就是数据接收的逻辑了。loopReader();} catch (Exception e) {failWebSocket(e, null);}}@Overridepublic void onFailure(Call call, IOException e) {failWebSocket(e, null);}});}
}​

2.4 数据收发准备

• HTTP 服务器的响应到达的时候，为数据收发做的准备，如：2.3代码onResponse;
• 共有6步：
• 第一步就是检查 HTTP 响应：
• 第二步：初始化用于输入输出的 Source 和 Sink。
• 第三步是调用回调 onOpen()。
• 第四步是初始化 Reader 和 Writer。
• 第五步是配置socket的超时时间为0，也就是阻塞IO。
• 第六步执行 loopReader()。

2.4.1 第一步就是检查 HTTP 响应：

• 根据 WebSocket 的协议，服务器端用如下响应，来表示接受建立 WebSocket 连接的请求：
• 1. 响应码是 101。
• 1. "Connection" header 的值为 "Upgrade"，以表明服务器并没有在处理完请求之后把连接个断开。
• 1. "Upgrade" header 的值为 "websocket"，以表明服务器接受后面使用 WebSocket 来通信。
• 1. "Sec-WebSocket-Accept" header 的值为，key + WebSocketProtocol.ACCEPT_MAGIC 做 SHA1 hash，然后做 base64 编码，来做服务器接受连接的验证。关于这部分的设计的详细信息，可参考 WebSocket 协议规范。

​
public final class RealWebSocket implements WebSocket, WebSocketReader.FrameCallback {void checkResponse(Response response) throws ProtocolException {//响应码是 101。if (response.code() != 101) {throw new ProtocolException("Expected HTTP 101 response but was '"+ response.code() + " " + response.message() + "'");}//"Connection" header 的值为 "Upgrade"，以表明服务器并没有在处理完请求之后把连接个断开。String headerConnection = response.header("Connection");if (!"Upgrade".equalsIgnoreCase(headerConnection)) {throw new ProtocolException("Expected 'Connection' header value 'Upgrade' but was '"+ headerConnection + "'");}//"Upgrade" header 的值为 "websocket"，以表明服务器接受后面使用 WebSocket 来通信。String headerUpgrade = response.header("Upgrade");if (!"websocket".equalsIgnoreCase(headerUpgrade)) {throw new ProtocolException("Expected 'Upgrade' header value 'websocket' but was '" + headerUpgrade + "'");}//"Sec-WebSocket-Accept" header 的值为，key + WebSocketProtocol.ACCEPT_MAGIC 做 SHA1 hash，然后做 base64 编码，来做服务器接受连接的验证。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 + "'");}}
}​

2.4.2 第二步：初始化用于输入输出的 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) {@Overridepublic void close() throws IOException {streamAllocation.streamFinished(true, streamAllocation.codec(), -1L, null);}};}
}​

2.4.3 第三步是调用回调 onOpen()

2.4.4 第四步是初始化 Reader 和 Writer：

• OkHttp使用 WebSocketReader 和 WebSocketWriter 来处理数据的收发。
• 在发送数据时将数据组织成帧，在接收数据时则进行反向操作，同时处理 WebSocket 的控制消息。
• WebSocket 的所有数据发送动作，都会在单线程线程池的线程中，通过 WebSocketWriter 执行。在这里会创建 ScheduledThreadPoolExecutor 用于跑数据的发送操作。
• WebSocket 协议中主要会传输两种类型的帧，一是控制帧，主要是用于连接保活的 Ping 帧等；二是用户数据载荷帧。
• 在这里会根据用户的配置，调度** Ping 帧**周期性地发送。
• 在调用 WebSocket 的接口发送数据时，数据并不是同步发送的，而是被放在了一个消息队列中。发送消息的 Runnable 从消息队列中读取数据发送。这里会检查消息队列中是否有数据，如果有的话，会调度发送消息的 Runnable 执行。

​
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.4.5第五步是配置socket的超时时间为0，也就是阻塞IO。

2.4.6第六步执行 loopReader()。这实际上是进入了消息读取循环了，也就是数据接收的逻辑了。

本次分享到此结束，感谢大家的阅读，觉得有所帮助的朋友点点关注点点赞！