Netty 笔记2

2021-09-14

十一、WebSocket 协议开发应用

WebSocket 是 HTML5 开始提供的一种浏览器与服务端进行全双工通信的网络技术，WebSocket 于2011年被 IETF 定为标准 RFC6455，WebSocket API 被W3C定为标准。

WebSocket 的特点：

单一的TCP连接，采用全双工模式通信；
对代理，防火墙和路由器透明；
无头部信息、Cookie 和身份验证；
无安全开销；
通过 ping/pong 帧保持链路激活；
服务器可以主动传消息给客户端，不再客户端需要轮询。

11.2、 WebSocket入门

WebSocket 设计出来的目的就是要取代轮询和Comet技术，使客户端浏览器具备像C/S 架构下的桌面系统一样实时通信的能力。

WebSocket 本质是就是一个TCP 连接，在数据传输的稳定性和数据量大小方面，和轮询以及Comet技术相比，具体很大的性能优势。

11.2.1、WebSocket 连接建立

建立WebSocket 连接时，需要通过客户端或浏览器向服务端发出握手请求。为建立一个 WebSocket 连接，客户端首先需要向服务器发起一个HTTP 请求，这个请求和通常的 HTTP 请求不同，包含了一些附加头信息，其中附加头信息Upgrade:WebSocket 表明这是一个申请协议升级的 HTTP 请求。如下图

11-1

请求消息中的 Sec-WebSocket-Key 是随机的，服务器会用这些数据来构造出一个 SHA-1的信息摘要，把 Sec-WebSocket-Key 加上一个魔幻字符串 258EAFA5-E914-47DA-95CA-C5AB0DC85B11，使用 SHA-1 加密，然后进行 BASE-64 编码，将结果作为 Sec-WebSocket-Accept 头的值，返回给客户端。服务端应答如下：

11-2

11.2.2、WebSocket 生命周期

握手成功后，客户端和服务端就可以通过message 的方式进行通信了，一个消息可以有一个或多个帧组成，WebSocket 的消息不一定对应一个特定网络层的帧，它可以被分割成多个帧或被合并。

11-3

11.2.3、WebSocket 连接关闭

为关闭 WebSocket 连接，客户端和服务端需要一个安全的方法关闭底层TCP 连接和 TLS回话。如果合适，丢弃任何可能已经接收的字节，必要时（受到攻击）可以通过任何手段关闭连接。

底层 TCP 连接，在正常情况下，应首先由服务器关闭。异常情况下，可由客户端发起 TCP Close，因此，当服务器被指示关闭 WebSocket 连接时，它应立即发起一个TCP Close 操作，客户端应该等待服务器的 TCP Close 。

11.3、Netty WebSocket 协议开发

Netty 基于 HTTP 协议栈开发了 WebSocket 协议栈，利用 Netty 的 WebSocket 的协议栈可以非常方便的开发出 WebSocket 的服务端和客户端。

下面将通过一个小例程来实现 Netty WebSocket 的客户端和服务端的开发，其主要功能为：客户端发送消息到服务端，服务端对请求消息进行判断，如果是合法的 WebSocket 消息，则响应返回 Netty WebSocket 服务端消息以及时间。

11.3.1、WebSocket 服务端开发

利用Netty 开发WebSocket 服务端的关键是网络 I/O 事件处理器handler，其中在 handler 处理的事件有:

判断区分 HTTP请求和 WebSocke 请求，因为第一次请求握手消息是由HTTP 协议承载；
如果是 HTTP 握手请求，则根据通过握手工厂类WebSocketServerHandshakerFactory 创建握手处理类WebSocketServerHandshaker 实例；其中握手处理了中的handshake 方法会将 websocket的编辑码类动态的添加到 ChannelPipeline
链路建立成功后，客户端提交文本到服务端，服务端分别判断，是否是关闭连接、是否为 ping 指令消息，是否是支持的文本消息，然后做出响应，返回给客户端。

代码如下：

public class WebSocketServer {

    public static void main(String[] arg) throws InterruptedException {
        WebSocketServer webSocketServer = new WebSocketServer();
        webSocketServer.bind(8082);
    }
    public void bind(int port) throws InterruptedException {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup,workGroup)
                    .channel(NioServerSocketChannel.class)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast("http-codec",new HttpServerCodec());
                            ch.pipeline().addLast("http-aggregator",new HttpObjectAggregator(65535));
                            ch.pipeline().addLast("http-chuncked",new ChunkedWriteHandler());
                            ch.pipeline().addLast("websocket-handler",new WebSocketHandler(port));
                        }
                    });
            Channel channel = serverBootstrap.bind(port).sync().channel();
            System.out.println("websocket sart at prot:"+port);
            System.out.println("please open your web browser and navigate to : http://localhost:"+port+"/");
            channel.closeFuture().sync();

        }finally {
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }

    public class WebSocketHandler extends SimpleChannelInboundHandler<Object>{

        private int port;
        // 处理 websocket 握手请求
        private WebSocketServerHandshaker handshaker;

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

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
            System.out.println("server receive msg:"+msg);
            if (msg instanceof FullHttpRequest){
                // 传统 Http 请求接入
                handlerHttpRequest(ctx,(FullHttpRequest)msg);
            }else if (msg instanceof WebSocketFrame){
                // websocket 接入
                handlerWebSocketFrame(ctx,(WebSocketFrame)msg);
            }
        }

        private void handlerHttpRequest(ChannelHandlerContext ctx, FullHttpRequest msg) {
            // Http 解码失败或者不是请求建立 websocket 连接的http 请求，返回请求失败
            if (!msg.decoderResult().isSuccess() || !"websocket".equals(msg.headers().get("Upgrade"))){
                sendHttpResponse(ctx,msg,new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,HttpResponseStatus.BAD_REQUEST));
                return;
            }
            // 构造握手请求响应
            WebSocketServerHandshakerFactory factory = new WebSocketServerHandshakerFactory("ws://localhost:"+port+"/websocket",
                    null,false);
            handshaker = factory.newHandshaker(msg);
            if (handshaker==null){
                WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
            }else {
                handshaker.handshake(ctx.channel(),msg);
            }
        }

        // 处理 websocket 帧
        private void handlerWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame msg) {
            // 判断是否为关闭 websocket 链路指令
            if (msg instanceof CloseWebSocketFrame){
                handshaker.close(ctx.channel(),((CloseWebSocketFrame) msg).retain());
                return;
            }
            // 判断是否为 ping 指令消息
            if (msg instanceof PingWebSocketFrame){
                ctx.channel().write(new PongWebSocketFrame(msg.content().retain()));
                return;
            }
            // 本例程仅支持文本消息，不支持二进制消息
            if (!(msg instanceof TextWebSocketFrame)){
                throw new UnsupportedOperationException(String.format("%s frame types ne supported.",msg.getClass().getName()));
            }
            // 返回应答消息
            String text = ((TextWebSocketFrame) msg).text();
            System.out.println(ctx.channel()+" recived client message: \r\n"+text);
            ctx.channel().writeAndFlush(new TextWebSocketFrame(text+",欢迎使用 Netty WebSocket 服务，现在时间："
                    + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))));

        }

        public void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, FullHttpResponse resp){
            // 返回应答给客户端
            if (resp.status().code()==200){
                ByteBuf buf = Unpooled.copiedBuffer(resp.status().code()+"", CharsetUtil.UTF_8);
                resp.content().writeBytes(buf);
                buf.release();
                HttpUtil.setContentLength(resp,resp.content().readableBytes());
            }
            ChannelFuture f = ctx.writeAndFlush(resp);
            // 如果非 keep-alive, 关闭连接
            if (!HttpUtil.isKeepAlive(req) || resp.status().code()!=200){
                f.addListener(ChannelFutureListener.CLOSE);
            }
        }

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

运行效果如下

11-4

11-5

11.4、Netty UDP 协议开发

用户数据报协议（User Datagram Protocol，UDP）是一种传输层协议。在 TCP/IP 网络中，它与 TCP 协议一样用于处理数据包，是一种无连接的协议。

TCP 协议在进行数据传输时，需要建立连接，并且每次传输的数据都需要进行确认。当不再进行传输数据时，还需要断开连接。这样做虽然安全，但是效率较低。而 UDP 协议正好避免了这些过程，它是一种没有复杂控制，提供面向无连接的通信服务协议。

UDP 协议具备以下特点：

没有各种连接：在传输数据前不需要建立连接，也避免了后续的断开连接。
不重新排序：对到达顺序混乱的数据包不进行重新排序。
没有确认：发送数据包无须等待对方确认。因此，使用 UDP 协议可以随时发送数据，但无法保证数据能否成功被目标主机接收。

11.4.1、UDP 服务端开发

public class NettyUdpServer {

    public static void main(String[] arg) throws InterruptedException {
        NettyUdpServer server = new NettyUdpServer();
        server.bind(8080);
    }

    public void bind(int port) throws InterruptedException {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        try {
            Bootstrap serverBootstrap = new Bootstrap();
            serverBootstrap.group(bossGroup)
                    // 指定通道类型为 NioDatagramChannel
                    .channel(NioDatagramChannel.class)
                    .option(ChannelOption.SO_BROADCAST,true)
                    //UDP相对于TCP不需要在客户端和服务端建立实际的连接，因此不需要为连接（ChannelPipeline）设置handler
                    .handler(new NettyUdpServerHandler());
            serverBootstrap.bind(port).sync().channel().closeFuture().await();

        }finally {
            bossGroup.shutdownGracefully();
        }
    }

    public class NettyUdpServerHandler extends SimpleChannelInboundHandler<DatagramPacket>{

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket msg) throws Exception {
            String content = msg.content().toString(CharsetUtil.UTF_8);
            System.out.println("收到客户端消息为："+content);
            if ("udp-req".equals(content)){
                String respStr = "欢迎使用UDP服务，盲盒开奖成功！";
                DatagramPacket respPacket = new DatagramPacket(Unpooled.copiedBuffer(respStr,CharsetUtil.UTF_8),msg.sender());
                ctx.writeAndFlush(respPacket);
            }
        }

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

11.4.2、UDP 客户端开发

public class NettyUdpClient {
    public static void main(String[] arg) throws InterruptedException {
        NettyUdpClient client = new NettyUdpClient();
        client.connect(8080);
    }

    public void connect(int port) throws InterruptedException {
        EventLoopGroup loopGroup = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(loopGroup)
                    .option(ChannelOption.SO_BROADCAST,true)
                    .channel(NioDatagramChannel.class)
                    .handler(new ClientUdpHandler());

            Channel channel = bootstrap.bind(8081).sync().channel();

            //向网段内的所有机器广播
            channel.writeAndFlush(new DatagramPacket(Unpooled.copiedBuffer("udp-req",CharsetUtil.UTF_8),
                    new InetSocketAddress("255.255.255.255",port)));
            //客户端等待15s用于接收服务端的应答消息，然后退出并释放资源
            if(!channel.closeFuture().await(15000)){
                System.out.println("请求超时！");
            }

        }finally {
            loopGroup.shutdownGracefully();
        }
    }

    private class ClientUdpHandler extends SimpleChannelInboundHandler<DatagramPacket>{

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket msg) throws Exception {
            String content = msg.content().toString(CharsetUtil.UTF_8);
            System.out.println("client receive msg is : "+content);
            ctx.close();
        }

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

十二、私有协议栈开发

广义上区分，通信协议可以分为共有协议和私有协议。由于私有协议的灵活性，和时候在公司或者组织内部使用，按需定制，升级方便，灵活性好。

绝大多数私有协议传输层基于TCP/IP，利用 Netty 的 NIO TCP 协议栈可以非常方便地进行私有协议的定制和开发。