Netty(五)——WebSocket之事

短命女 2022-05-15 18:36 199阅读 0赞

在说WebSocket以前，我们再来看看HTTP协议，HTTP1.0，HTTP1.1，HTTP2.0每个版本的更新带来更高效的更丰富的功能：短连接、长连接、缓存处理的丰富、断点续传、错误通知的丰富、多路复用、请求优先级、header压缩……更多的我们来看这篇文章：[HTTP1.0、HTTP1.1和HTTP2.0的区别][HTTP1.0_HTTP1.1_HTTP2.0] （[https://www.cnblogs.com/zhangyfr/p/8662673.html）][https_www.cnblogs.com_zhangyfr_p_8662673.html]

HTTP解决了我们很多问题，而且也非常强大，但是它有哪些不足之处呢？我们来看下：1，HTTP协议为半双工协议，可以在 client和server两个方向上传输，但是不能同时传输，意味着同一时刻，只能一个方向上传送；    2，HTTP消息冗长而复杂，包含消息头、消息体、换行符，虽然功能强大，但是在某些方面也成了不足。              例如我们需要server向client端推送数据利用HTTP就很难，例如一些监控系统，随着后台数据的变化，前边也跟着相应的变化，随后后来的轮询机制等可以解决这个问题，但是这样采用了长连接，大量消耗了服务器的宽带和资源。也就是在client和server期望实时通讯的情况下，Http就有些头疼了。亦是HTML5定义了WebSocket协议，更好的节省服务器资源和宽带，并达到实时通讯的目的。好下边我们来看WebSocket的相关内容：

WebSocket，浏览器和服务器只需要做一个握手动作（非HTTP三次握手），然后之间就会形成一条快速通道，两者就可以相互传送数据了，是基于TCP双向全双工进行消息传递，其特点包括：1，单一的TCP连接，采用全双工模式通信；    2，对代理、防火墙和路由器透明；     3，无头部信息、Cookie和身份验证；     4，无安全开销；     5，通过“ping/pong”帧保持链路激活；     6，服务器可以主动传递消息给客户端，不再需要客户端轮询。

好了，讲了WebSocket，我们来看下，利用Netty如何方便快速的开发WebSocket的服务端，下边通过一个实时传输文本的例子来看，服务端代码：

public class WebSocketServer {
    
    
        public void run(final int port) {
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workerGroup = new NioEventLoopGroup();
    
            try {
                ServerBootstrap b = new ServerBootstrap();
                b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast("http-codec",new HttpServerCodec());
                        pipeline.addLast("aggregator",new HttpObjectAggregator(65536));
                        pipeline.addLast("http-chunked",new ChunkedWriteHandler());
                        //业务处理
                        ch.pipeline().addLast("handler", new WebSocketServerHandler());
                    }
                });
                Channel ch =b.bind(port).sync().channel();
                System.out.println("web socket server started at port" + port + ".");
                System.out.println("open your browser and navigate to http://localhost:"  + port + "/");
                ch.closeFuture().sync();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    
    
        public static void main(String[] args) {
            int port = 8091;
            if (args != null && args.length > 0) {
                try {
                    port = Integer.valueOf(args[0]);
                } catch (NumberFormatException e) {
    
                }
            }
    
            new WebSocketServer().run(port);
        }
    }
    
    
    
    
    public class WebSocketServerHandler extends SimpleChannelInboundHandler<Object> {
    
        private static final Logger logger = Logger.getLogger(WebSocketServerHandler.class.getName());
    
        private WebSocketServerHandshaker handshaker;
    
        /**
         * 接受到消息处理
         * @param ctx
         * @param msg
         * @throws Exception
         */
        @Override
        protected void messageReceived(ChannelHandlerContext ctx, Object msg) throws Exception {
            if(msg instanceof FullHttpRequest){
                //传统的http接入
                handlerHttpRequest(ctx,(FullHttpRequest)msg);
            }else if(msg instanceof WebSocketFrame){
                //WebSocket接入
                handlerWebSocketFrame(ctx,(WebSocketFrame)msg);
            }
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            ctx.flush();
        }
    
    
        /**
         * 处理正常请求
         * @param ctx
         * @param req
         */
        private void handlerHttpRequest(ChannelHandlerContext ctx,FullHttpRequest req){
            //如果HTTP解码失败，返回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:8091/websocket",null,false);
            handshaker = wsFactory.newHandshaker(req);
            if(handshaker ==null){
                WebSocketServerHandshakerFactory .sendUnsupportedVersionResponse(ctx.channel());
            }else{
                handshaker.handshake(ctx.channel(),req);
            }
        }
    
    
        /**
         * 处理websocket请求
         * @param ctx
         * @param frame
         */
        private void handlerWebSocketFrame(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 supported",frame.getClass().getName()));
            }
    
            //返回应答消息
            String request = ((TextWebSocketFrame) frame).text();
            if(logger.isLoggable(Level.FINE)){
                logger.fine(String.format("%s received %s",ctx.channel(),request));
            }
            System.out.println("server 接受到的信息：" + request);
            ctx.channel().write(new TextWebSocketFrame(request + " ,欢迎使用netty websocket服务，现在时刻：" + new Date().toString()));
        }
    
        /**
         * 响应客户端消息
         * @param ctx
         * @param req
         * @param res
         */
        private static void sendHttpResponse(ChannelHandlerContext ctx , FullHttpRequest req, FullHttpResponse res){
            if(res.status().code() !=200){
                ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8);
                res.content().writeBytes(buf);
                buf.release();
                HttpHeaderUtil.setContentLength(res,res.content().readableBytes());
            }
    
            ChannelFuture f = ctx.channel().writeAndFlush(res);
            if(!HttpHeaderUtil.isKeepAlive(req) || res.status().code()!=200){
                f.addListener(ChannelFutureListener.CLOSE);
            }
        }
    
    }

前端页面：

<html>
        <head>
            <meta charset="UTF-8">
            Netty WebSocket 时间服务器
        </head>
        <br>
        <body>
            <br>
            <script type="text/javascript">
                var  socket;
                if(!window.WebSocket){
                    window.WebSocket = window.MozWebSocket;
                }
                if(window.WebSocket){
                    socket = new WebSocket("ws://localhost:8091/websocket");
                    socket.onmessage = function (event) {
                        var ta =document.getElementById('responseText');
                        ta.value ="";
                        ta.value = event.data;
                    };
                    socket.onopen =function (event) {
                        var ta =document.getElementById("responseText");
                        ta.value  = "打开WebSocket服务正常，浏览器支持websocket!"
                    };
                    socket.onclose = function (event) {
                        var ta =document.getElementById("responseText");
                        ta.value="";
                        ta.value="WebSocket关闭！";
                    };
                }else{
                    alert("sorry,您的浏览器不支持websocket协议！");
                }
    
                function send(message) {
                    if(!window.WebSocket){
                        return;
                    }
                    if(socket.readyState = WebSocket.OPEN){
                        socket.send(message);
                        socket.send("ljh study netty")
                    }else{
                        alert("WebSocket连接没有建立成功！");
                    }
                }
            </script>
    
            <form onsubmit="return false;">
                <input type="text" name="message" value="Netty最佳实践"/>
                <br><br>
                <input type="button" value="发送WebSocket消息" onclick="send(this.form.message.value)"/>
                <hr color="blue"/>
                <h3>服务端返回的应答消息</h3>
                <textarea id="responseText" style="width: 500px;height :300px;"></textarea>
    
            </form>
        </body>
    </html>

Netty对WebSocket协议栈的支持还是非常棒的，更多的我们需要了解可以到WebSocket[官网][Link 1] （[http://www.websocket.org/index.html）][http_www.websocket.org_index.html]

[HTTP1.0_HTTP1.1_HTTP2.0]: https://www.cnblogs.com/zhangyfr/p/8662673.html
[https_www.cnblogs.com_zhangyfr_p_8662673.html]: https://www.cnblogs.com/zhangyfr/p/8662673.html%EF%BC%89
[Link 1]: http://www.websocket.org/index.html
[http_www.websocket.org_index.html]: http://www.websocket.org/index.html%EF%BC%89