Netty之通信协议和私有协议栈开发(三) 之 WebSocket协议开发

77 阅读 0 评论 51 点赞

我是靠谱客的博主义气火，最近开发中收集的这篇文章主要介绍Netty之通信协议和私有协议栈开发(三) 之 WebSocket协议开发，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

2021SC@SDUSC

WebSocket协议开发

一. HTTP协议的弊端及 WebSocket协议的特点

二. Websocket 连接过程 ,生命周期, 连接关闭

三.Netty中WebSocket协议栈的开发

四.小结

一. HTTP协议的弊端及 WebSocket协议的特点

这次分析的内容和我们最近正在学习的计算机网络的传输层有关联，所以理解起来没有那么抽象了。

为什么这次我们要分析WebSocket呢？前两次博客我们都是在分析如何使用Netty进行HTTP协议开发，但是，Http协议的开销问题，导致它们不适用于低延迟的应用。为了解决这些问题，我们引入了WebSocket。

WebSocket将网络套接字，也就是我们常说的socket，引入到了客户端和服务端。浏览器和服务器之间可以通过socket建立持久的连接，双方随时都可以互发数据给对方，而不是之前由客户端控制的一请求一应答模式。

我们先来总结一下HTTP协议的弊端：

1.HTTP协议是半双工的协议。通俗的解释就是，当对方在说话时，你就不能说话了，也就是说一个时间点上，只能一个人在说话。
2.HTTP协议冗长而繁琐。HTTP消息包含消息头，消息体，换行符等，采用文本形式传播，相比二进制通信协议，非常冗长且繁琐。
3.针对服务器推送的黑客攻击。例如长时间轮询。

针对HTTP协议的弊端再来看一下WebSocket协议的特点：

1.单一的TCP连接，采用全双工通信。
2.对代理，防火墙，路由器透明。
3.无头部信息，cookie和身份验证。
4.无安全开销。
5.通过ping帧保持链路激活。
6.服务器可以主动的对客户端进行发送消息，而不需要轮询。

总的来说我们WebSocket被设计出来的目的就是为了取代轮询这种高消耗的方式。

二. Websocket 连接过程 ,生命周期, 连接关闭

建立连接过程：

1.客户端浏览器首先要向服务器发送一个HTTP请求，这个请求和通常的HTTP请求不同，包含了一些附加信息，其中附加信息“UpgradeWebSocket”表明这是一个申请协议升级的http请求。
2.服务器解析这些附加头信息，然后生成应答信息返回给客户端，这样客户端和服务器端的webSocket连接就建立起来了
3.这个连接会一直持续到客户端或者服务器的某一方主动关闭连接。

生命周期：

在握手成功后，客户端和服务器就可以通过 "message" 方式进行通信了。

连接关闭：

如果要关闭websocket连接，客户端和服务器需要通过一个安全的方法关闭底层TCP连接及TLS会话。如果合适，丢弃任何可能接收的字节。必要时可以通过任何可用的手段关闭连接。

三.Netty中WebSocket协议栈的开发

Netty基于HTTP协议栈开发了WebSocket协议栈,利用Netty 的WebSocket 协议栈可以非常方便地开发出 WebSocket客户端和服务端。这一部分我主要通过一个Netty服务端实例的开发，来分析如何使用Netty进行WebSocket开发。

首先学习WebSocket 服务端的功能。WebSocket服务端接收到请求消息之后，先对消息的类型进行判断，如果不是WebSocket 握手请求消息，则返回 HTTP 400BAD REQUEST响应给客户端。

客户端的握手请求消息如下：

GET /chat HTTP/1.1
Host: server.example.com
Upgrade:websocket
Connection:Upgrade
Sec-WeSocket-Key:dGhlweidxlkjsdZQ== 
Origin:http://example.com
Sec-WebSocket-Protocol:chat,superchat
Sec-webSocket-Version:13

服务端对握手请求消息进行处理，构造握手响应返回，双方的Socket连接正式建立，服务端返回的握手应答消息如下：

HTTP/1.1 101 Switching Protocols
Upgrade:websocket
Connection:Upgrade
Sec-WebSocket-accept:siefjosigdfsl=
Sec-WebSocket-Protocol：chat

连接建立成功后，到被关闭之前，双方都可以主动向对方发送消息，这点跟HTTP的一请求一应答模式存在很大的差别。相比于HTTP，它的网络利用率更高，可以通过全双工的方式进行消息发送和接收。

这里浏览器和服务器只需要做一个握手的动作，然后，浏览器和服务器之间就形成了一条快速通道。两者之间就直接可以数据互相传送。

先来看WebSocket服务端代码实现：

public class NettyWebSocketServer {

    private int port;

    public NettyWebSocketServer(int port){
        this.port = port;
    }

    public void start(){
        // boss 是处理客户端连接的线程池
        // worker 是处理从客户端连接转发过来的IO数据的读写线程池
        NioEventLoopGroup boss = new NioEventLoopGroup();
        NioEventLoopGroup worker = new NioEventLoopGroup();
        try{
            // ServerBootstrap 对外一个便利创建服务端，Builder建造者设计模式
            ServerBootstrap sb = new ServerBootstrap();
            // 绑定线程池
            sb.group(boss,worker)
                    // 绑定channel 服务端绑定NioServerSocketChannel,此实现jdk的ServerSocketChannel
                    .channel(NioServerSocketChannel.class)
                    // 绑定服务端相关参数，可添加绑定多个参数
                    .option(ChannelOption.SO_BACKLOG, 1024) //指定此套接口排队的最大连接个数
                    // IO事件处理类，主要处理IO事件的读写
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            // 请求和应答消息编码或者解码为HTTP消息
                            pipeline.addLast(new HttpServerCodec());
                            // 将HTTP消息的多个部分组合成一条完整的HTTP消息
                            pipeline.addLast(new HttpObjectAggregator(65535));
                            // 向客户端发送HTML5文件，主要用于支持浏览器和服务端进行WebSocket通信
                            pipeline.addLast(new ChunkedWriteHandler());
                            pipeline.addLast(new NettyWebSocketServerHandler());

                        }
                    });
            // 绑定端口，同步等待成功
            ChannelFuture cf = sb.bind(port).sync();
            System.out.println("服务已启动.................监听端口：" + port);
            // 等待服务器监听端口关闭
            cf.channel().closeFuture().sync();
        }catch (Exception e){
            // 优雅关闭线程资源
            boss.shutdownGracefully();
            worker.shutdownGracefully();
        }
    }

    public static void main(String[] args) {
        NettyWebSocketServer nettyServer = new NettyWebSocketServer(8080);
        nettyServer.start();
    }
}

关键是看向ChannelPipeline添加了什么handler。

首先添加HttpServerCodec，将请求和应答消息编码或者解码为HTTP消息。

然后添加HttpObjectAggregator，它的目的是将HTTP消息的多个部分组合成一条完整的HTTP消息。

然后添加ChunkedWriteHandler，来向客户端发送HTMLS文件,它主要用于支持浏览器和服务端进行WebSocket通信。

最后增加WebSocket服务端handler。

看到这里会感觉WebSocket的服务启动类和HTTP协议的非常类似。我们没有看到在ChannelPipeline中增加 WebSocket的 Handler，那是如何处理WebSocket消息？下面详细来看WebSocketServerHandler的处理逻辑。

public class NettyWebSocketServerHandler extends ChannelInboundHandlerAdapter {

    private WebSocketServerHandshaker handshaker;

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 客户端第一次接入，升级Upgrade  websocket
        if(msg instanceof FullHttpRequest){
            handleHttpRequest(ctx, (FullHttpRequest)msg);
        }
        // websocket数据交互
        else if(msg instanceof WebSocketFrame){
            handleWebSocketFrame(ctx , (WebSocketFrame)msg);
        }
    }

    private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
        // 判断是否是关闭链路的指令
        if(frame instanceof CloseWebSocketFrame){
            handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
            return;
        }
        // 是否是Ping消息
        if(frame instanceof PingWebSocketFrame){
            ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
            return;
        }
        // 当前需求仅需要文本消息
        if(!(frame instanceof TextWebSocketFrame)){
            throw new UnsupportedOperationException(String.format("%s frame types not supportes", frame.getClass().getName()));
        }
        // 返回应答消息
        String request = ((TextWebSocketFrame) frame).text();
        System.out.println("当前收到的消息是： " + request);
        String timeStr = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());
        ctx.channel().write(new TextWebSocketFrame(request+ ", netty webSocket 服务端，time now is " + timeStr));
    }

    private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) {
        // 如果http解析失败，返回异常
        if(!req.decoderResult().isSuccess() || (!"websocket".equals(req.headers().get("Upgrade")))){
            sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
            return;
        }
        // 构造握手响应返回
        WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory("ws://localhost:8080/websocket",null, false);
        handshaker = wsFactory.newHandshaker(req);
        if(handshaker == null){
            WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
        }else {
            handshaker.handshake(ctx.channel(),req);
        }
    }

    private void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, DefaultFullHttpResponse response) {
        // 返回给客户端
        if(response.status().code() != 200){
            ByteBuf buf = Unpooled.copiedBuffer(req.toString().toString(), CharsetUtil.UTF_8);
            response.content().writeBytes(buf);
            buf.release();
            response.headers().set("Content-Length",response.content().readableBytes());
        }
        // 如果是非Keep-Alive,关闭连接
        ChannelFuture f = ctx.channel().writeAndFlush(response);
        if(!req.headers().get("Connection").equals("keep-alive") || response.status().code()!=200){
            f.addListener(ChannelFutureListener.CLOSE);
        }
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

关键代码详细解读：

 // 客户端第一次接入，升级Upgrade  websocket
        if(msg instanceof FullHttpRequest){
            handleHttpRequest(ctx, (FullHttpRequest)msg);
        }

首先来看第一次握手请求消息由 HTTP协议承载，所以它是一个HTTP消息，执行handleHttpRequest方法来处理WebSocket 握手请求。

        // 如果http解析失败，返回异常
        if(!req.decoderResult().isSuccess() || (!"websocket".equals(req.headers().get("Upgrade")))){
            sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
            return;
        }

然后对握手请求消息进行判断，如果消息头中没有包含Upgrade字段或者它的值不是 websocket，则返回HTTP 400响应。

// 构造握手响应返回
        WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory("ws://localhost:8080/websocket",null, false);
        handshaker = wsFactory.newHandshaker(req);
        if(handshaker == null){
            WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
        }else {
            handshaker.handshake(ctx.channel(),req);
        }

握手请求简单校验通过之后，构造握手工厂，创建握手处理类WebSocketServerHandshaker。在这里会通过它构造握手响应消息返回给客户端，同时将WebSocket相关的编码 WebSocket Encoder 和解码类WebSocket Decoder动态添加到ChannelPipeline中，用于WebSocket消息的编解码。
添加 WebSocket Encoder 和 WebSocket Decoder之后，服务端就可以自动对 WebSocket消息进行编解码了，后面的业务 handler可以直接对WebSocket对象进行操作。

下面继续分析链路建立成功之后的操作:

客户端提交请求消息给服务端，WebSocketServerHandler 接收到的是已经解码后的WebSocketFrame消息。

这里会调用 handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame)方法对WebSocket请求消息进行处理，首先需要对控制帧进行判断，如果是关闭链路的控制消息，就调用WebSocketServerHandshaker 的close方法关闭WebSocket连接;如果是维持链路的Ping 消息，则构造Pong 消息返回。

最后获取请求消息字符串，对它处理后通过构造新的息返回给客户端，由于握手应答时动态增加了新的编码类，所以，可以直接发送对象。客户端浏览器接收到服务端的应答消息后，将其内容取出展示到浏览器页面中。

四.小结

这一部分主要是分析了基于Netty的Websocket协议开发。这里最重要的一下代码已经指出来了。比如客户端第一次接入，升级为websocket。创建握手处理类WebSocketServerHandshaker，并且动态的添加编解码器，让后面的业务 handler可以直接对WebSocket对象进行操作等。

学习到目前我感觉这其中最关键的是handler的编写，这是体现核心逻辑的一部分。学习完Websocket，我感觉除了通过握手可以建立一个双方同时可以通信的全双工信道外，比较重要的也就是是这个消息接收后handler的一步步处理的逻辑。