Netty搭建简单WebSocket通信

短命女 2021-11-11 12:02 384阅读 0赞

### 概述 ###

websocket使用主从线程组模型，仅实现浏览器发送数据，服务器接收数据控制台输出并转发给所有连接的客户端

端口绑定8088

### 代码 ###

主启动类 WSServer.java

package com.imooc.netty.websocket;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    
    /**
     * @Author Sakura
     * @Date 5/8/2019
     **/
    
    
    public class WSServer {
    
        public static void main(String[] args) throws Exception {
    
            //创建主从线程组
            EventLoopGroup mainGroup = new NioEventLoopGroup();
            EventLoopGroup subGroup = new NioEventLoopGroup();
    
            try {
                ServerBootstrap serverBootstrap = new ServerBootstrap();
    
                serverBootstrap.group(mainGroup,subGroup)
                               .channel(NioServerSocketChannel.class)
                               .childHandler(new WSServerInitializer());
    
                //绑定端口，同步启动
                ChannelFuture channelFuture = serverBootstrap.bind(8088).sync();
    
                //监听关闭，同步启动
                channelFuture.channel().closeFuture().sync();
            } finally {
                mainGroup.shutdownGracefully();
                subGroup.shutdownGracefully();
            }
        }
    }

初始化器 WSServerInitializer.java

package com.imooc.netty.websocket;
    
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelPipeline;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.handler.codec.http.HttpObjectAggregator;
    import io.netty.handler.codec.http.HttpServerCodec;
    import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
    import io.netty.handler.stream.ChunkedWriteHandler;
    
    /**
     * @Author Sakura
     * @Date 5/8/2019
     **/
    public class WSServerInitializer extends ChannelInitializer<SocketChannel> {
        protected void initChannel(SocketChannel socketChannel) throws Exception {
    
            ChannelPipeline pipeline = socketChannel.pipeline();
    
            //websoket基于http，所以有http编码解码器
            pipeline.addLast(new HttpServerCodec());
            //对写大数据流的支持
            pipeline.addLast(new ChunkedWriteHandler());
            //对httpMessage进行聚合，聚合成HttpRequest或HttpResponse
            //几乎在Netty的编程都会使用到此handler
            pipeline.addLast(new HttpObjectAggregator(1024*64));
    
    
            //====================以上是用于支持Http协议===========================
            /**
             * websocket服务器处理协议，用于指定给客户端访问的路由：/ws (比如  ws://localhost:8088/ws)
             * 本handler会处理以下繁重的事务
             * 握手动作：handshaking(close,ping--请求,pong--响应)
             * websocket都是以frames进行传输，不同数据类型frames也不同
             */
            pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
            //自定义通信助手类，客户端消息处理与消息回送
            pipeline.addLast(new ChatHandler());
    
    
        }
    }

自定义助手类 ChatHandler.java

package com.imooc.netty.websocket;
    
    import io.netty.channel.Channel;
    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.websocketx.TextWebSocketFrame;
    import io.netty.util.concurrent.GlobalEventExecutor;
    
    import java.time.LocalDateTime;
    
    /**
     * @Author Sakura
     * @Date 5/8/2019
     * 处理消息的handler
     * TextWebSocketFrame 在netty中专门为websocket处理文本的对象，frame是消息载体
     **/
    public class ChatHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
    
        //用于记录和管理所有客户端channel的ChannelGroup
        private static ChannelGroup clients = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame msg) throws Exception {
    
            //获取客户端传输的消息
            String content = msg.text();
            System.out.println("接受到的数据为：" + content);
    
    //        for (Channel channel:clients) {
    //            channel.writeAndFlush(
    //                    new TextWebSocketFrame("[服务器于"+ LocalDateTime.now()+ "收到消息]：" + content)
    //            );
    //        }
            //该方法同以上循环
            clients.writeAndFlush(new TextWebSocketFrame("[服务器于"+ LocalDateTime.now()+ "收到消息]：" + content));
        }
    
        /**
         *客户端连接服务端后(打开链接)
         * 获取客户端channel，并放入ChannelGroup中管理
         */
    
        @Override
        public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
            clients.add(ctx.channel());
        }
    
        /**
         * 当触发handlerRemoved方法时，ChannelGroup会自动移除对应客户端的channel
         * 所以无需手动remove
         */
    
        @Override
        public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
            //clients.remove(ctx.channel());
            //每个channel都有标是自身的id，分长短。有大型channel时短id可能重复
            System.out.println("客户端断开，channel对应长id为："+ ctx.channel().id().asLongText());
            System.out.println("客户端断开，channel对应短id为："+ ctx.channel().id().asShortText());
        }
    }

前端代码

<!DOCTYPE html>
    <html>
    	<head>
    		<meta charset="utf-8" />
    		<title></title>
    	</head>
    	<body>
    		<text>消息发送</text> <br/>
    		<input type="text" id="msgSend" />
    		<input type="button" value="发送" onclick="CHAT.chat()"/><br/>
    		<text>消息接收</text><br/>
    		<div id="msgRecv" style="background-color: blanchedalmond;"></div>
    		
    		<script type="text/javascript">
    			window.CHAT = {
    				socket:null,
    				init:function(){
    					if(window.WebSocket){
    						
    						CHAT.socket = new WebSocket("ws://127.0.0.1:8088/ws");
    						CHAT.socket.onopen=function(){
    							console.log("建立连接");
    						},
    						CHAT.socket.onclose=function(){
    							console.log("连接关闭")
    						},
    						CHAT.socket.onerror=function(){
    							console.log("发生错误")
    						},
    						CHAT.socket.onmessage=function(e){
    							console.log("收到消息:"+e.data)
    							
    							var msgRecv = document.getElementById("msgRecv");
    							var html = msgRecv.innerHTML;
    							msgRecv.innerHTML = html + "<br/>" + e.data;
    						}
    					}else{
    						alert("该浏览器不支持websocket");
    					}
    				},
    	
    				chat:function(){
    					var msg = document.getElementById("msgSend");
    					CHAT.socket.send(msg.value)
    				}
    			}
    			
    			CHAT.init()
    		</script>
    	</body>
    </html>

### 测试 ###

项目启动后，打开前端网页，打开开发者工具

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjA4OTE3NQ_size_16_color_FFFFFF_t_70][]

传输一条数据后

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjA4OTE3NQ_size_16_color_FFFFFF_t_70 1][]

关闭网页后控制台

![20190805201554447.png][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjA4OTE3NQ_size_16_color_FFFFFF_t_70]: /images/20211109/0ae7f422de714940bb0385cdba938c18.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjA4OTE3NQ_size_16_color_FFFFFF_t_70 1]: /images/20211109/fe3ca93954b743628e43e2318795b9b9.png
[20190805201554447.png]: /images/20211109/bdc95d121ee94b0785cb6d1d0de8b6c8.png