Spring Boot+Netty+Websocket实现后台向前端推送信息

àì夳堔傛蜴生んèń 2023-10-11 19:49 23阅读 0赞

Netty 是一个利用 Java 的高级网络的能力，隐藏其背后的复杂性而提供一个易于使用的**API的客户端/服务器框架**。

可能在此之前你没有接触过，不过不要担心，下面我们通过一个消息推送的例子来看一下netty是怎么使用的。

## 1.添加Maven依赖 ##

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

## 2.设置主启动类 ##

import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    
    @SpringBootApplication
    public class MessageSendApplication {
        
        public static void main(String[] args) {
        
            SpringApplication.run(MessageSendApplication.class, args);
            try {
        
                new NettyServer(12345).start();
                System.out.println("https://blog.csdn.net/moshowgame");
                System.out.println("http://127.0.0.1:6688/netty-websocket/index");
            } catch (Exception e) {
        
                System.out.println("NettyServerError:" + e.getMessage());
            }
    
        }
    }

## 3.NettyServer ##

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;
    
    /**
     * NettyServer Netty服务器配置
     */
    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", null, true, 65536 * 10));
                                ch.pipeline().addLast(new WebSocketHandler());
                            }
                        });
                ChannelFuture cf = sb.bind().sync(); // 服务器异步创建绑定
                System.out.println(NettyServer.class + " 启动正在监听： " + cf.channel().localAddress());
                cf.channel().closeFuture().sync(); // 关闭服务器通道
            } finally {
        
                group.shutdownGracefully().sync(); // 释放线程池资源
                bossGroup.shutdownGracefully().sync();
            }
        }
    }

## 4.MyChannelHandlerPool ##

通道组池，管理所有websocket连接

import io.netty.channel.group.ChannelGroup;
    import io.netty.channel.group.DefaultChannelGroup;
    import io.netty.util.concurrent.GlobalEventExecutor;
    
    /**
     * MyChannelHandlerPool
     * 通道组池，管理所有websocket连接
     */
    public class MyChannelHandlerPool {
        
    
        public MyChannelHandlerPool(){
        }
    
        public static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    
    }

## 5.WebSocketHandler ##

处理ws以下几种情况：

*  channelActive与客户端建立连接
 *  channelInactive与客户端断开连接
 *  channelRead0客户端发送消息处理

import cn.hutool.json.JSONObject;
    import cn.hutool.json.JSONUtil;
    import com.example.messagesend.netty.NettyConfig;
    import io.netty.channel.ChannelHandler;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
    import io.netty.util.AttributeKey;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.stereotype.Component;
    
    @Component
    @ChannelHandler.Sharable
    @Slf4j
    public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
        
    
        /**
         * 一旦连接，第一个被执行
         */
        @Override
        public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        
            log.info("有新的客户端链接：[{}]", ctx.channel().id().asLongText());
            // 添加到channelGroup 通道组
            NettyConfig.getChannelGroup().add(ctx.channel());
        }
    
        /**
         * 读取数据
         */
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        
            log.info("服务器收到消息：{}", msg.text());
            // 获取用户ID,关联channel
            String text = msg.text();
            String uid = text.split(":")[0];
            String message = text.split(":")[1];
    //        JSONObject jsonObject = JSONUtil.parseObj(msg.text());
            NettyConfig.getChannelMap().put(uid, ctx.channel());
    
            // 将用户ID作为自定义属性加入到channel中，方便随时channel中获取用户ID
            AttributeKey<String> key = AttributeKey.valueOf("userId");
            ctx.channel().attr(key).setIfAbsent(uid);
    
            // 回复消息
            ctx.channel().writeAndFlush(new TextWebSocketFrame(uid+"："+message));
        }
    
        @Override
        public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        
            log.info("用户下线了:{}", ctx.channel().id().asLongText());
            // 删除通道
            NettyConfig.getChannelGroup().remove(ctx.channel());
            removeUserId(ctx);
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        
            log.info("异常：{}", cause.getMessage());
            // 删除通道
            NettyConfig.getChannelGroup().remove(ctx.channel());
            removeUserId(ctx);
            ctx.close();
        }
    
        /**
         * 删除用户与channel的对应关系
         */
        private void removeUserId(ChannelHandlerContext ctx) {
        
            AttributeKey<String> key = AttributeKey.valueOf("userId");
            String userId = ctx.channel().attr(key).get();
            NettyConfig.getChannelMap().remove(userId);
        }
    }

## 6.Netty配置 ##

管理全局Channel以及用户对应的channel（推送消息）

import io.netty.channel.Channel;
    import io.netty.channel.group.ChannelGroup;
    import io.netty.channel.group.DefaultChannelGroup;
    import io.netty.util.concurrent.GlobalEventExecutor;
    
    import java.util.concurrent.ConcurrentHashMap;
    
    public class NettyConfig {
        
        /**
         * 定义全局单利channel组 管理所有channel
         */
        private static volatile ChannelGroup channelGroup = null;
    
        /**
         * 存放请求ID与channel的对应关系
         */
        private static volatile ConcurrentHashMap<String, Channel> channelMap = null;
    
        /**
         * 定义两把锁
         */
        private static final Object lock1 = new Object();
        private static final Object lock2 = new Object();
    
    
        public static ChannelGroup getChannelGroup() {
        
            if (null == channelGroup) {
        
                synchronized (lock1) {
        
                    if (null == channelGroup) {
        
                        channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
                    }
                }
            }
            return channelGroup;
        }
    
        public static ConcurrentHashMap<String, Channel> getChannelMap() {
        
            if (null == channelMap) {
        
                synchronized (lock2) {
        
                    if (null == channelMap) {
        
                        channelMap = new ConcurrentHashMap<>();
                    }
                }
            }
            return channelMap;
        }
    
        public static Channel getChannel(String userId) {
        
            if (null == channelMap) {
        
                return getChannelMap().get(userId);
            }
            return channelMap.get(userId);
        }
    }

## 7.socket.html ##

<!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>

## 8.Controller ##

写好了html当然还需要一个controller来引导页面。

import cn.hutool.core.util.RandomUtil;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.PathVariable;
    import org.springframework.web.bind.annotation.RestController;
    import org.springframework.web.servlet.ModelAndView;
    
    @RestController
    public class TestController {
        
        @GetMapping("/index")
        public ModelAndView index() {
        
            ModelAndView mav = new ModelAndView("socket");
            mav.addObject("uid", RandomUtil.randomNumbers(6));
            return mav;
        }
    }

下面我们打开页面测试一下  
![在这里插入图片描述][60f240f040f045bb80020ddd09503097.png_pic_center]

## 9.改造netty支持url参数 ##

我们重新写一个WebSocketHandler:**MyWebSocketHandler**，然后调整加载handler的顺序,优先MyWebSocketHandler在WebSocketServerProtocolHandler之上。

改造**MyWebSocketHandler**的**channelRead**方法，首次连接会是一个**FullHttpRequest**类型，可以通过**FullHttpRequest.uri()**获取完整ws的URL地址，之后接受信息的话，会是一个**TextWebSocketFrame**类型。

MyWebSocketHandler:

import com.alibaba.fastjson.JSON;
    import com.example.messagesend.netty.MyChannelHandlerPool;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    import io.netty.handler.codec.http.FullHttpRequest;
    import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
    
    import java.util.HashMap;
    import java.util.Map;
    
    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
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        
            //首次连接是FullHttpRequest，处理参数 by zhengkai.blog.csdn.net
            if (null != msg && msg instanceof FullHttpRequest) {
        
                FullHttpRequest request = (FullHttpRequest) msg;
                String uri = request.uri();
    
                Map paramMap=getUrlParams(uri);
                System.out.println("接收到的参数是："+ JSON.toJSONString(paramMap));
                //如果url包含参数，需要处理
                if(uri.contains("?")){
        
                    String newUri=uri.substring(0,uri.indexOf("?"));
                    System.out.println(newUri);
                    request.setUri(newUri);
                }
    
            }else if(msg instanceof TextWebSocketFrame){
        
                //正常的TEXT消息类型
                TextWebSocketFrame frame=(TextWebSocketFrame)msg;
                System.out.println("客户端收到服务器数据：" +frame.text());
                sendAllMessage(frame.text());
            }
            super.channelRead(ctx, msg);
        }
    
        @Override
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame textWebSocketFrame) throws Exception {
        
    
        }
    
        private void sendAllMessage(String message){
        
            //收到信息后，群发给所有channel
            MyChannelHandlerPool.channelGroup.writeAndFlush( new TextWebSocketFrame(message));
        }
    
        private static Map getUrlParams(String url){
        
            Map<String,String> map = new HashMap<>();
            url = url.replace("?",";");
            if (!url.contains(";")){
        
                return map;
            }
            if (url.split(";").length > 0){
        
                String[] arr = url.split(";")[1].split("&");
                for (String s : arr){
        
                    String key = s.split("=")[0];
                    String value = s.split("=")[1];
                    map.put(key,value);
                }
                return  map;
    
            }else{
        
                return map;
            }
        }
    }

然后在 NettyServer中调整顺序：  
![在这里插入图片描述][f2fb81cec42c4e04bda64aa9eadd5b53.png_pic_center]

我们html中的socket连接地址也需要改一下：

socket = new WebSocket("ws://127.0.0.1:12345/ws?uid=666&gid=777");

接着重新运行一下，看看我们控制台的效果：  
![在这里插入图片描述][a2702122a187485fb8bdae568b29e910.png_pic_center]

[60f240f040f045bb80020ddd09503097.png_pic_center]: https://img-blog.csdnimg.cn/60f240f040f045bb80020ddd09503097.png#pic_center
[f2fb81cec42c4e04bda64aa9eadd5b53.png_pic_center]: https://img-blog.csdnimg.cn/f2fb81cec42c4e04bda64aa9eadd5b53.png#pic_center
[a2702122a187485fb8bdae568b29e910.png_pic_center]: https://img-blog.csdnimg.cn/a2702122a187485fb8bdae568b29e910.png#pic_center