netty实现websocket

痛定思痛。 2021-11-09 14:28 424阅读 0赞

1.搭建springboot web 项目，参考

[https://blog.csdn.net/u013184307/article/details/98057621][https_blog.csdn.net_u013184307_article_details_98057621]

2.导入包

<dependency>
       <groupId>io.netty</groupId>
       <artifactId>netty-all</artifactId>
       <version>4.1.36.Final</version>
    </dependency>

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMxODQzMDc_size_16_color_FFFFFF_t_70][]

3.启动器中需要new一个NettyServer，并显式调用启动netty

package com.horse.red;
    
    import com.horse.red.netty.NettyServer;
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    
    @SpringBootApplication
    public class RedApplication {
    
       public static void main(String[] args) {
          SpringApplication.run(RedApplication.class, args);
    
          try {
             new NettyServer(12345).start();
          } catch (Exception e) {
             e.printStackTrace();
          }
       }
    
    }

package com.horse.red.netty;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    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;
    
    public class NettyServer {
    
        private final int port;
    
        public NettyServer(int port) {
            this.port = port;
        }
    
        public void start() throws Exception {
            EventLoopGroup bossGroup = new NioEventLoopGroup();
    
            EventLoopGroup group = new NioEventLoopGroup();
            try {
                ServerBootstrap sb = new ServerBootstrap();
                sb.option(ChannelOption.SO_BACKLOG, 1024);
                sb.group(group, bossGroup) // 绑定线程池
                        .channel(NioServerSocketChannel.class) // 指定使用的channel
                        .localAddress(this.port)// 绑定监听端口
                        .childHandler(new ChannelInitializer<SocketChannel>() { // 绑定客户端连接时候触发操作
    
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
                                System.out.println("收到新连接");
                                //websocket协议本身是基于http协议的，所以这边也要使用http解编码器
                                ch.pipeline().addLast(new HttpServerCodec());
                                //以块的方式来写的处理器
                                ch.pipeline().addLast(new ChunkedWriteHandler());
                                ch.pipeline().addLast(new HttpObjectAggregator(8192));
                                ch.pipeline().addLast(new WebSocketServerProtocolHandler("/ws", "WebSocket", true, 65536 * 10));
                                ch.pipeline().addLast(new MyWebSocketHandler());
                            }
                        });
                ChannelFuture cf = sb.bind().sync(); // 服务器异步创建绑定
                System.out.println(NettyServer.class + " 启动正在监听： " + cf.channel().localAddress());
                cf.channel().closeFuture().sync(); // 关闭服务器通道
            } finally {
                group.shutdownGracefully().sync(); // 释放线程池资源
                bossGroup.shutdownGracefully().sync();
            }
        }
    }

package com.horse.red.netty;
    
    import io.netty.channel.group.ChannelGroup;
    import io.netty.channel.group.DefaultChannelGroup;
    import io.netty.util.concurrent.GlobalEventExecutor;
    
    public class MyChannelHandlerPool {
    
        public MyChannelHandlerPool(){}
    
        public static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    
    }

package com.horse.red.netty;
    
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
    
    public class MyWebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            System.out.println("与客户端建立连接，通道开启！");
    
            //添加到channelGroup通道组
            MyChannelHandlerPool.channelGroup.add(ctx.channel());
        }
    
        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            System.out.println("与客户端断开连接，通道关闭！");
            //添加到channelGroup 通道组
            MyChannelHandlerPool.channelGroup.remove(ctx.channel());
        }
    
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
            System.out.println("客户端收到服务器数据：" + msg.text());
            sendAllMessage(msg.text());
        }
    
        private void sendAllMessage(String message){
            //收到信息后，群发给所有channel
            MyChannelHandlerPool.channelGroup.writeAndFlush( new TextWebSocketFrame(message));
        }
    }

<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN" "http://www.w3.org/TR/xhtml1/DTD/xhtml1transitional.dtd">
    <html xmlns="http://www.w3.org/1999/xhtml">
    <head>
        <meta http-equiv="Content-Type" content="text/html; charset=utf-8" />
        <title>Netty-Websocket</title>
        <script type="text/javascript">
            // by zhengkai.blog.csdn.net
            var socket;
            if(!window.WebSocket){
                window.WebSocket = window.MozWebSocket;
            }
            if(window.WebSocket){
                socket = new WebSocket("ws://127.0.0.1:12345/ws");
                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 = "Netty-WebSocket服务器。。。。。。连接  \r\n";
                };
                socket.onclose = function(event){
                    var ta = document.getElementById('responseText');
                    ta.value = "Netty-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;">
        <label>ID</label><input type="text" name="uid" value="${uid!!}" /> <br />
        <label>TEXT</label><input type="text" name="message" value="这里输入消息" /> <br />
        <br /> <input type="button" value="发送ws消息"
                      onClick="send(this.form.uid.value+':'+this.form.message.value)" />
        <hr color="black" />
        <h3>服务端返回的应答消息</h3>
        <textarea id="responseText" style="width: 1024px;height: 300px;"></textarea>
    </form>
    </body>
    </html>

运行效果

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMxODQzMDc_size_16_color_FFFFFF_t_70 1][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMxODQzMDc_size_16_color_FFFFFF_t_70 2][]

上面为群发，没有指明发给指定人，以下解决

当有客户端连接时候会被`channelActive`监听到，当断开时会被`channelInactive`监听到，一般在这两个方法中去保存/移除客户端的通道信息，而通道信息保存在`ChannelSupervise`中：

ChannelGroup是netty提供用于管理web于服务器建立的通道channel的，其本质是一个高度封装的set集合，在服务器广播消息时，可以直接通过它的writeAndFlush将消息发送给集合中的所有通道中去。但在查找某一个客户端的通道时候比较坑爹，必须通过channelId对象去查找，而channelId不能人为创建，所有必须通过map将channelId的字符串和channel保存起来。

package com.horse.red.netty;
    
    import io.netty.channel.Channel;
    import io.netty.channel.ChannelId;
    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.util.concurrent.ConcurrentHashMap;
    import java.util.concurrent.ConcurrentMap;
    
    public class MyChannelHandlerPool {
    
        public MyChannelHandlerPool(){}
    
        public static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    
        private  static ConcurrentMap<String, ChannelId> ChannelMap=new ConcurrentHashMap();
    
        public  static void addChannel(Channel channel){
            channelGroup.add(channel);
            ChannelMap.put(channel.id().asShortText(),channel.id());
        }
        public static void removeChannel(Channel channel){
            channelGroup.remove(channel);
            ChannelMap.remove(channel.id().asShortText());
        }
        public static  Channel findChannel(String id){
            return channelGroup.find(ChannelMap.get(id));
        }
    }

@Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        System.out.println("客户端收到服务器数据：" + msg.text());
        ChannelId channelId = ctx.channel().id();
        //sendAllMessage(msg.text());
        sendAllMessage2(msg.text(),channelId);
    }
    
    private void sendAllMessage(String message){
        //收到信息后，群发给所有channel
        MyChannelHandlerPool.channelGroup.writeAndFlush( new TextWebSocketFrame(message));
    
    }
    
    private void sendAllMessage2(String message,ChannelId channelId){
        //收到信息后，发给指定channel
        MyChannelHandlerPool.findChannel(channelId.asShortText()).writeAndFlush( new TextWebSocketFrame(message));
    
    }

@Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("与客户端建立连接，通道开启！");
    
        //添加到channelGroup通道组
        MyChannelHandlerPool.addChannel(ctx.channel());
    }
    
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("与客户端断开连接，通道关闭！");
        //添加到channelGroup 通道组
        MyChannelHandlerPool.removeChannel(ctx.channel());
    }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMxODQzMDc_size_16_color_FFFFFF_t_70 3][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMxODQzMDc_size_16_color_FFFFFF_t_70 4][]

完美解决！！！！！！！！！！

[https_blog.csdn.net_u013184307_article_details_98057621]: https://blog.csdn.net/u013184307/article/details/98057621
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMxODQzMDc_size_16_color_FFFFFF_t_70]: /images/20211109/da1dc7d7a3c6474b919411090b6918e6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMxODQzMDc_size_16_color_FFFFFF_t_70 1]: /images/20211109/4b95fa8aaf944ccb80052203e263b378.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMxODQzMDc_size_16_color_FFFFFF_t_70 2]: /images/20211109/ca3fc1868fd049d3b3c86118fdede2a4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMxODQzMDc_size_16_color_FFFFFF_t_70 3]: /images/20211109/eaff5c3a6814421392545929d07e0926.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMxODQzMDc_size_16_color_FFFFFF_t_70 4]: /images/20211109/7f26292880b04452addb178f59a9633a.png