SpringBoot2+netty+webSocket实现前后端互相发消息

你的名字 2023-05-22 10:24 64阅读 0赞

## 1.Maven需要的依赖 ##

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

## 初始化netty设置解码器,编码器（可以自定义） ##

public class WebScoketServerInitialzer  extends ChannelInitializer<SocketChannel> {
    
        @Override
        protected void initChannel(SocketChannel socketChannel) throws Exception {
            ChannelPipeline pipeline = socketChannel.pipeline();
    
            //http解码器（websocket是基于http协议的，所以可以直接用现成的
            http解码器）
            pipeline.addLast(new HttpServerCodec());
            //对写大数据流的支持
            pipeline.addLast(new ChunkedWriteHandler());
            //设置单次请求的文件的大小
            pipeline.addLast(new HttpObjectAggregator(1024*1024*10));
            //webscoket 服务器处理的协议，用于指定给客户端连接访问的路由 :/ws
            pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
            //自定义handler(作用类似controller,客户端和服务器端之间发消息都在这个自定义handler里面)
            pipeline.addLast("handler",new WebSocketHandler());
    
            
    
            }
    }

## websocket启动类的实现 ##

@Component
    public class WebSocketServer {
    
        private static EventExecutorGroup group = new DefaultEventExecutorGroup(1024);
    
        public void run(String... args) throws Exception {
    
            //创建两个线程
            EventLoopGroup bossGroup = null;
            EventLoopGroup workerGroup = null;
            try {
                bossGroup = new NioEventLoopGroup();
                workerGroup = new NioEventLoopGroup();
                //服务对象
                ServerBootstrap serverBootstrap = new ServerBootstrap();
                //构建工程线程池
                serverBootstrap.group(bossGroup, workerGroup)
                        //Nio长连接
                        .channel(NioServerSocketChannel.class)
                        .childOption(ChannelOption.SO_KEEPALIVE, true)
                        .option(ChannelOption.SO_BACKLOG, 1024)
                        .handler(new LoggingHandler(LogLevel.INFO))
                        //初始化 过滤 解码
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
                                ChannelPipeline pipeline = ch.pipeline();
                                //配置http解码组件
                                pipeline.addLast("http-codec", new HttpServerCodec());
                                pipeline.addLast("ping", new IdleStateHandler(2, 1, 3, TimeUnit.MINUTES));
                                //把多个消息转换为一个单一 Http请求或响应
                                pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
                                //文件传输
                                pipeline.addLast("http-chunked", new ChunkedWriteHandler());
                                //逻辑  连接 服务ing 异常 断开
                                pipeline.addLast(group, "handler", new WebSocketHandler());
                            }
                        });
                //绑定监听端口号
                ChannelFuture channelFuture = serverBootstrap.bind(8082).sync();
                //关闭
                channelFuture.channel().closeFuture().sync();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                if (bossGroup != null) {
                    bossGroup.shutdownGracefully();
                }
    
                if (workerGroup != null) {
                    workerGroup.shutdownGracefully();
                }
            }
        }

## websocket处理类 ##

@Slf4j
    public class WebSocketHandler extends ChannelInboundHandlerAdapter {
        private WebSocketServerHandshaker handShaker;
        Connections connections = Connections.getInstance();
    
        /**
         * 打开连接
         *
         */
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            log.info("------------页面建立连接---------------");
            Connections connections = Connections.getInstance();
            connections.saveConnection("1",ctx);
            super.channelActive(ctx);
        }
        /**
         * 断开连接时 移除管道 删除map中管道对应的用户
         *
         * @param ctx
         * @throws Exception
         */
        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            super.channelInactive(ctx);
            log.info("[" + ctx.channel().remoteAddress().toString() + "] connect lost.");
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            log.info("-----------连接出错------------");
            cause.printStackTrace();
            ctx.close();
        }
    
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            // 传统的HTTP接入
            if (msg instanceof FullHttpRequest) {
                handleHttpRequest(ctx, (FullHttpRequest) msg);
                //websocket建立连接是基于websocket
                log.info("这是http接入了");
            }
            // WebSocket接入
            else if (msg instanceof WebSocketFrame) {
                handleWebSocketFrame(ctx, (WebSocketFrame) msg);
            }
        }
    
        /**
         * 判断请求内容
         *
         * @param ctx
         * @param frame
         */
        private void handleWebSocketFrame(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()));
            }
    
            try {
    
                log.info("收到的客户端消息是："+ ((TextWebSocketFrame) frame).text());
                ctx.channel().write(new TextWebSocketFrame("我收到消息啦"));
                ctx.flush();
            } catch (Exception e2) {
                e2.printStackTrace();
                log.info("处理Socket请求异常");
    
            }
    
    
        }
    
        private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) throws Exception {
    
            // 如果HTTP解码失败，返回HTTP异常
            if (!req.decoderResult().isSuccess() || (!"websocket".equals(req.headers().get("Upgrade")))) {
                sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, BAD_REQUEST));
                return;
            }
    
            String uri = req.uri();
            System.out.println("请求的链接地址" + uri);
    
            // 构造握手响应返回
            WebSocketServerHandshakerFactory wsFactory =
                    new WebSocketServerHandshakerFactory("ws://" + req.headers().get(HttpHeaderNames.HOST) + uri, null, false);
            handShaker = wsFactory.newHandshaker(req);
            if (handShaker == null) {
                WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
            } else {
    
                handShaker.handshake(ctx.channel(), req);
            }
        }
    
        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();
                HttpUtil.setContentLength(res, res.content().readableBytes());
            }
    
            // 如果是非Keep-Alive，关闭连接
            ChannelFuture f = ctx.channel().writeAndFlush(res);
            if (!HttpUtil.isKeepAlive(req) || res.status().code() != 200) {
                f.addListener(ChannelFutureListener.CLOSE);
            }
        }

## 配置Application启动类 ##

@SpringBootApplication
    @EnableSwagger2
    public class LoginApplication implements CommandLineRunner {
    
        public static void main(String[] args) {
            SpringApplication.run(LoginApplication.class, args);
        }
    
        WorkThreadPool workThreadPool = new WorkThreadPool();
        public void run(String... args) throws Exception {
            workThreadPool.getPool().execute(() -> {
                try {
                    new WebSocketServer().run();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
    
    
        }
    }

客户端在线测试工具：[websocket在线测试工具][websocket]  
1.启动SpringBoot项目  
2.输入连接地址连接  
![在这里插入图片描述][20200428150907242.png]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0x5bm5lWm9l_size_16_color_FFFFFF_t_70]  
可以测试发消息给客户端  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0x5bm5lWm9l_size_16_color_FFFFFF_t_70 1]  
在代码里面服务端也可以打印客户端的消息  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0x5bm5lWm9l_size_16_color_FFFFFF_t_70 2]  
![在这里插入图片描述][20200428151231741.png]  
这样就可以通信了，获取到客户端消息。  
如果需要群发消息，需要自己手写前端代码建立连接的时候传一个唯一标识，后端用HashMap或者自定义一个类用来存储来自不同地方的连接，发消息时候遍历取到每一个连接，使用ChannelHandlerContext.channel.write(new TextWebsocketFrame(“消息内容”))；就可以实现群发消息  
还有就是，如果业务需要返回前端的数据来自数据库（要调用service层代码情况），实例化的时候有可能使用Autowire注解实例化失败会报空指针（先检查Handler类有没有用@Component），试着换用setter方式注入，就可以解决了.  
[源码地址][Link 1]

[websocket]: http://oktools.net/websocket
[20200428150907242.png]: /images/20230521/0e388075ffec4f97a1497c40ad161555.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0x5bm5lWm9l_size_16_color_FFFFFF_t_70]: /images/20230521/6f690ab9ce8a40ba9016419d168a02c8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0x5bm5lWm9l_size_16_color_FFFFFF_t_70 1]: /images/20230521/35c778b1407b4dce8f030a8e0d9cc400.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0x5bm5lWm9l_size_16_color_FFFFFF_t_70 2]: /images/20230521/ae8801f057f849a1a3d0a8114ace4095.png
[20200428151231741.png]: /images/20230521/c540b0e192024b43bd96873cad9dcd8d.png
[Link 1]: https://gitee.com/LynneLing/Utils.git