Netty之WebSocket

喜欢ヅ旅行 2023-06-20 06:19 21阅读 0赞

### WebSocket 简介 ###

WebSocket 协议最大特点就是，服务器可以主动向客户端推送信息，客户端也可以主动向服务器发送信息，是真正的双向平等对话，属于[服务器推送技术][Link 1]的一种。

### WebSocket的通信建立——握手过程 ###

WebSocket的握手使用HTTP来实现，客户端发送带有Upgrade头的HTTP Request消息。服务端根据请求，做Response。

### 请求报文： ###

GET wss://www.example.cn/webSocket HTTP/1.1
    Host: www.example.cn
    Connection: Upgrade
    Upgrade: websocket
    Sec-WebSocket-Version: 13
    Origin: http://example.cn
    Sec-WebSocket-Key: afmbhhBRQuwCLmnWDRWHxw==
    Sec-WebSocket-Protocol: chat, superchat
    Sec-WebSocket-Extensions: permessage-deflate; client_max_window_bits
    复制代码

详细解释一下：

*  **第1、2行**：与HTTP的Request的请求行一样，这里使用的是HTTPS协议，所以对应的是wss请求。
 *  **第3行**：Connection：HTTP1.1中规定Upgrade只能应用在直接连接中。带有Upgrade头的HTTP1.1消息必须含有Connection头，因为Connection头的意义就是，任何接收到此消息的人（往往是代理服务器）都要在转发此消息之前处理掉Connection中指定的域（即不转发Upgrade域）。
 *  **第4行**：Upgrade是HTTP1.1中用于定义转换协议的header域。 如果服务器支持的话，客户端希望使用已经建立好的HTTP（TCP）连接，切换到WebSocket协议。
 *  **第5行**：Sec-WebSocket-Version标识了客户端支持的WebSocket协议的版本列表。
 *  **第6行**：Origin为安全使用，防止跨站攻击，浏览器一般会使用这个来标识原始域。
 *  **第7行**：Sec-WebSocket-Key是一个Base64encode的值，这个是客户端随机生成的，用于服务端的验证，服务器会使用此字段组装成另一个key值放在握手返回信息里发送客户端。
 *  **第8行**：Sec\_WebSocket-Protocol是一个用户定义的字符串，用来区分同URL下，不同的服务所需要的协议，标识了客户端支持的子协议的列表。
 *  **第9行**：Sec-WebSocket-Extensions是客户端用来与服务端协商扩展协议的字段，permessage-deflate表示协商是否使用传输数据压缩，client\_max\_window\_bits表示采用LZ77压缩算法时，滑动窗口相关的SIZE大小。

注：如果对压缩扩展协商的细节感兴趣，可参考下面的RFC7692了解更多细节。 [Compression Extensions for WebSocket RFC7692][]

### 响应报文： ###

HTTP/1.1 101
    Server: nginx/1.12.2
    Date: Sat, 11 Aug 2018 13:21:27 GMT
    Connection: upgrade
    Upgrade: websocket
    Sec-WebSocket-Accept: sLMyWetYOwus23qJyUD/fa1hztc=
    Sec-WebSocket-Protocol: chat
    Sec-WebSocket-Extensions: permessage-deflate;client_max_window_bits=15
    复制代码

详细解释一下：

*  **第1行**：HTTP的版本为HTTP1.1，返回码是101，开始解析Header域（不区分大小写）。
 *  **第2,3行**：服务器信息与时间。
 *  **第4行**：Connection字段，包含Upgrade。
 *  **第5行**：Upgrade字段，包含websocket。
 *  **第6行**：Sec-WebSocket-Accept字段，详细介绍一下：

> Sec-WebSocket-Accept字段生成步骤：
> 
> 1.  将Sec-WebSocket-Key与协议中已定义的一个GUID “258EAFA5-E914-47DA-95CA-C5AB0DC85B11”进行拼接。
> 2.  将步骤1中生成的字符串进行SHA1编码。
> 3.  将步骤2中生成的字符串进行Base64编码。

> 客户端通过验证服务端返回的Sec-WebSocket-Accept的值, 来确定两件事情:

> 1.  服务端是否理解WebSocket协议, 如果服务端不理解,那么它就不会返回正确的Sec-WebSocket-Accept，则建立WebSocket连接失败。
> 2.  服务端返回的Response是对于客户端的此次请求的,而不是之前的缓存。 主要是防止有些缓存服务器返回缓存的Response.

*  **第7行**：Sec-WebSocket-Protocol字段，要判断是否之前的Request握手带有此协议，如果没有，则连接失败。
 *  **第8行**：扩展协议协商，支持压缩，且LZZ的滑动窗口大小为15。

至此，握手过程就完成了，此时的TCP连接不会释放。客户端和服务端可以互相通信了。

### WebSocket基于Netty的代码实现 ###

MyWebSocketServerHandler

package com.yj.websocket.service;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.Channel;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelFutureListener;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    import io.netty.channel.group.ChannelGroup;
    import io.netty.channel.group.DefaultChannelGroup;
    import io.netty.handler.codec.http.DefaultFullHttpResponse;
    import io.netty.handler.codec.http.FullHttpRequest;
    import io.netty.handler.codec.http.HttpResponseStatus;
    import io.netty.handler.codec.http.HttpVersion;
    import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
    import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
    import io.netty.handler.codec.http.websocketx.WebSocketFrame;
    import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
    import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
    import io.netty.util.CharsetUtil;
    import io.netty.util.concurrent.GlobalEventExecutor;
    
    public class MyWebSocketServerHandler extends SimpleChannelInboundHandler<Object> {
    	
    	private WebSocketServerHandshaker handshaker;
    	
    	public static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    	
    	private static final String WEB_SOCKET_URL = "ws://localhost:8899/websocket";
    
    	@Override
    	public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    		ctx.flush();
    	}
    
    	private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) {
    		if (!req.getDecoderResult().isSuccess() || (!"websocket".equals(req.headers().get("Upgrade")))) {
    			sendHttpResponse(ctx, req,
    					new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
    			return;
    		}
    		WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(
    				WEB_SOCKET_URL, null, false);
    
    		handshaker = wsFactory.newHandshaker(req);
    
    		if (handshaker == null) {
    			WebSocketServerHandshakerFactory.sendUnsupportedWebSocketVersionResponse(ctx.channel());
    		} else {
    			handshaker.handshake(ctx.channel(), req);
    		}
    	}
    
    	private void handlerWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
    		// 判断是否关闭链路的指令
    
    		if (frame instanceof CloseWebSocketFrame) {
    			handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
    			return;
    		}
    		Channel incoming = ctx.channel();
    		String request = ((TextWebSocketFrame) frame).text();
    		System.out.println("服务端收到客户[" + incoming.id().asShortText() + "]的消息:" + request);
    		for (Channel channel : channels) {
    			if (channel != incoming) {
    				channel.writeAndFlush(new TextWebSocketFrame("[" + incoming.id().asShortText() + "]" + request));
    			} else {
    				channel.writeAndFlush(new TextWebSocketFrame("[you]" + request));
    			}
    		}
    	}
    
    	private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, DefaultFullHttpResponse res) {
    		// 返回应答给客户端
    		if (res.getStatus().code() != 200) {
    			ByteBuf buf = Unpooled.copiedBuffer(res.getStatus().toString(), CharsetUtil.UTF_8);
    			res.content().writeBytes(buf);
    			buf.release();
    		}
    
    		// 如果是非Keep-Alive，关闭连接
    		ChannelFuture f = ctx.channel().writeAndFlush(res);
    		if (!isKeepAlive(req) || res.getStatus().code() != 200) {
    			f.addListener(ChannelFutureListener.CLOSE);
    		}
    	}
    
    	private static boolean isKeepAlive(FullHttpRequest req) {
    		return false;
    	}
    
    	@Override
    	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    		cause.printStackTrace();
    		ctx.close();
    
    	}
    
    	@Override
    	public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
    		Channel incoming = ctx.channel();
    		channels.add(incoming);
    		channels.writeAndFlush(new TextWebSocketFrame("[SERVER] - " + incoming.id().asShortText() + " 加入"));
    		System.out.println("用户:" + incoming.id().asShortText() + "加入");
    	}
    
    	@Override
    	public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
    		Channel incoming = ctx.channel();
    		channels.remove(incoming);
    		channels.writeAndFlush(new TextWebSocketFrame("[SERVER] - " + incoming.id().asShortText() + " 离开"));
    		System.out.println("用户:" + incoming.id().asShortText() + "离开");
    	}
    
    	@Override
    	protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
    		if (msg instanceof FullHttpRequest) {
    			handleHttpRequest(ctx, ((FullHttpRequest) msg));
    		} else if (msg instanceof WebSocketFrame) {
    			handlerWebSocketFrame(ctx, (WebSocketFrame) msg);
    		}
    	}
    }

核心处理流程

>  *  web发起一次类似是http的请求，并在`channelRead0`方法中进行处理，并通过instanceof去判断帧对象是`FullHttpRequest`还是`WebSocketFrame`，建立连接是时候会是`FullHttpRequest`
>  *  在`handleHttpRequest`方法中去创建websocket，首先是判断`Upgrade`是不是websocket协议，若不是则通过`sendHttpResponse`将错误信息返回给客户端，紧接着通过`WebSocketServerHandshakerFactory`创建socket对象并通过handshaker握手创建连接
>  *  在连接创建好后的所以消息流动都是以`WebSocketFrame`来体现
>  *  在`handlerWebSocketFrame`去处理消息，也可能是客户端发起的关闭指令，ping指令等等

MyWebSocketChannelHandler

package com.yj.websocket.service;
    
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.handler.codec.http.HttpObjectAggregator;
    import io.netty.handler.codec.http.HttpServerCodec;
    import io.netty.handler.stream.ChunkedWriteHandler;
    
    public class MyWebSocketChannelHandler extends ChannelInitializer<SocketChannel>{
    
    	@Override
    	protected void initChannel(SocketChannel e) throws Exception {
    		e.pipeline().addLast("http-codec",new HttpServerCodec());
    		e.pipeline().addLast("aggregator",new HttpObjectAggregator(65536));
    		e.pipeline().addLast("http-chunked",new ChunkedWriteHandler());
    		e.pipeline().addLast("handler",new MyWebSocketServerHandler());
    	}
    }

index.html

<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN" "http://www.w3.org/TR/xhtml1/DTD/xhtml1-transitional.dtd">
    <html xmlns="http://www.w3.org/1999/xhtml">
    	<head>
    		<meta http-equiv="Content-Type" content="text/html; charset=utf-8" />
    		<title>
    			WebSocket客户端
    		</title>
    	<script type="text/javascript">
    		var socket;
    
    		if (!window.WebSocket) {
    
    			window.WebSocket = window.MozWebSocket;
    
    		}
    
    		if (window.WebSocket) {
    
    			socket = new WebSocket("ws://localhost:8899/websocket");
    
    			socket.onmessage = function(event) {
    
    				var ta = document.getElementById('responseText');
    
    				ta.value += event.data + "\r\n";
    
    			};
    
    			socket.onopen = function(event) {
    
    				var ta = document.getElementById('responseText');
    
    				ta.value = "打开WebSoket服务正常，浏览器支持WebSoket!" + "\r\n";
    
    			};
    
    			socket.onclose = function(event) {
    
    				var ta = document.getElementById('responseText');
    
    				ta.value = "";
    
    				ta.value = "WebSocket 关闭" + "\r\n";
    
    			};
    
    		} else {
    
    			alert("您的浏览器不支持WebSocket协议！");
    
    		}
    
    		function send(message) {
    
    			if (!window.WebSocket) {
    				return;
    			}
    			if (socket.readyState == WebSocket.OPEN) {
    
    				socket.send(message);
    
    			} else {
    
    				alert("WebSocket 连接没有建立成功！");
    
    			}
    
    		}
    	</script>
    	</head>
    	<body>
    		<form onSubmit="return false;">
    			<input type="text" name="message" value="Hello WebSocket" />
    			<br/>
    			<br/>
    			<input type="button" value="发送消息" onClick="send(this.form.message.value)"
    			/>
    			<hr color="blue" />
    			<h3>
    				服务端返回的应答消息
    			</h3>
    			<textarea id="responseText" style="width: 1024px;height: 300px;">
    			</textarea>
    		</form>
    	</body>
    
    </html>

启动服务端，打开浏览器访问[http://127.0.0.1:8080/index][http_127.0.0.1_8080_index]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MzEzNjM1_size_16_color_FFFFFF_t_70][]

服务端显示

用户:69f84395加入

客户端发送消息

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MzEzNjM1_size_16_color_FFFFFF_t_70 1][]

服务端收到消息

服务端收到客户[69f84395]的消息:Hello WebSoket

[Link 1]: https://en.wikipedia.org/wiki/Push_technology
[Compression Extensions for WebSocket RFC7692]: https://tools.ietf.org/pdf/rfc7692.pdf
[http_127.0.0.1_8080_index]: http://127.0.0.1:8080/index
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MzEzNjM1_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191208172547972.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MzEzNjM1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MzEzNjM1_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20191208172637658.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MzEzNjM1,size_16,color_FFFFFF,t_70