介绍 Netty WebSocket 整合 SpringBoot 的方式，源码

项目准备：

引入依赖：

导入 Netty 的依赖到 Pom 文件中

  1. <dependency>
  2.   <groupId>io.netty</groupId>
  3.   <artifactId>netty-all</artifactId>
  4.   <version>4.1.73.Final</version>
  5. </dependency>

创建服务端：

使用单例模式创建服务端，确保获取到的是同一个服务端对象

  1. package com.dmbjz.netty;
  3. import io.netty.bootstrap.ServerBootstrap;
  4. import io.netty.channel.ChannelFuture;
  5. import io.netty.channel.EventLoopGroup;
  6. import io.netty.channel.nio.NioEventLoopGroup;
  7. import io.netty.channel.socket.nio.NioServerSocketChannel;
  8. import lombok.extern.slf4j.Slf4j;
  9. import org.springframework.stereotype.Component;
  12. @Component
  13. @Slf4j
  14. public class WsServer {
  16.     //创建单例模式
  17.     private static class SingletionWsServer{
  18.         static final WsServer instance = new WsServer();
  19.     }
  21.     //返回单例模式对象
  22.     public static WsServer getInstance(){
  23.         return SingletionWsServer.instance;
  24.     }
  27.     private EventLoopGroup bossGroup;
  28.     private EventLoopGroup workGroup;
  29.     private ServerBootstrap bootstrap;
  30.     private ChannelFuture channelFuture;
  32.     /* 构造方法内进行初始化 */
  33.     public WsServer(){
  34.         bossGroup = new NioEventLoopGroup();
  35.         workGroup = new NioEventLoopGroup();
  36.         bootstrap = new ServerBootstrap();
  37.         bootstrap.group(bossGroup,workGroup)
  38.             .channel(NioServerSocketChannel.class)
  39.             .childHandler(new WsServerInitializer());
  40.     }
  43.     public void start(){
  44.         //额外线程启动Netty而不是在Main方法中启动，因此无需异步
  45.         //使用Spring容器进行托管，也无需进行关闭
  46.         this.channelFuture = bootstrap.bind(8088);
  47.         log.warn("Netty WebSocket Server启动完毕...");
  48.     }
  51. }

添加ServerInitializer：

添加 WebSocket 的初始化器

  1. package com.dmbjz.netty;
  3. import io.netty.channel.ChannelInitializer;
  4. import io.netty.channel.ChannelPipeline;
  5. import io.netty.channel.socket.SocketChannel;
  6. import io.netty.handler.codec.http.HttpObjectAggregator;
  7. import io.netty.handler.codec.http.HttpServerCodec;
  8. import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
  9. import io.netty.handler.stream.ChunkedWriteHandler;
  12. /* 自定义WebSocket初始化器 */
  13. public class WsServerInitializer extends ChannelInitializer<SocketChannel> {
  16.     @Override
  17.     protected void initChannel(SocketChannel ch) throws Exception {
  19.         ChannelPipeline pipeline = ch.pipeline();
  21.         //Http的编码解码器
  22.         pipeline.addLast(new HttpServerCodec());
  24.         //WebSocket以块的方式写数据，需要添加ChunkedWrite处理器
  25.         pipeline.addLast(new ChunkedWriteHandler());
  27.         /*
  28.          *  Http数据在传输过程中是分段的，HttpObjectAggregator可以将多个段聚合起来
  29.          *      所以浏览器发送大量数据时就会发出多次Http请求
  30.          */
  31.         pipeline.addLast(new HttpObjectAggregator(1024*64));
  33.         /* WebSocket的数据是以帧的形式传递，需要指定 WebSocketServerProtocolHandler和自定义处理帧数据Handler
  34.          *   WebSocketServerProtocolHandler的功能是将Http协议升级为WebSocket协议来保持长连接，并解析 websocket 访问地址
  35.          *   浏览器访问WebSocket的形式是  ws://localhost:7070/访问路径,参数就为 “/访问路径”
  36.          *       例如访问的是 ws://localhost:7070/rog16hp，这里的handler就要指定为 "/rog16hp"
  37.          */
  38.         pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
  40.         pipeline.addLast(new ChatHandler());
  42.     }
  45. }

添加消息处理Handler：

  1. package com.dmbjz.netty;
  3. import io.netty.channel.Channel;
  4. import io.netty.channel.ChannelHandlerContext;
  5. import io.netty.channel.SimpleChannelInboundHandler;
  6. import io.netty.channel.group.ChannelGroup;
  7. import io.netty.channel.group.DefaultChannelGroup;
  8. import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
  9. import io.netty.util.concurrent.GlobalEventExecutor;
  10. import lombok.extern.slf4j.Slf4j;
  12. import java.time.LocalDateTime;
  15. /* 处理消息的Handler */
  16. @Slf4j
  17. public class ChatHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
  20.     //用于记录和管理所有客户端的Channel，可以进行Channel的自动移除，因此不需要再写 handlerRemoved 方法
  21.     private static ChannelGroup clients = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
  24.     @Override
  25.     protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
  27.         String content = msg.text();
  28.         log.info("服务器收到消息:"+content);
  30.         //对所有加入到Group的channel回复消息
  31.         clients.forEach(channel ->
  32.                 channel.writeAndFlush(new TextWebSocketFrame("[服务器接收到消息:]" +LocalDateTime.now()+",消息为:"+ content))
  33.         );
  35.     }
  38.     /*
  39.      * 当客户端连接服务端后触发
  40.      * 获取客户端的Channel，并且放到ChannelGrou中去进行管理
  41.      */
  42.     @Override
  43.     public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
  44.         Channel channel = ctx.channel();
  45.         clients.add(channel);
  46.     }
  49.     @Override
  50.     public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
  51.         log.warn("客户端断开，长ID为:"+ctx.channel().id().asLongText());
  52.         log.warn("客户端断开，短ID为:"+ctx.channel().id().asShortText());
  53.     }
  56.     /* 客户端发生异常后触发 */
  57.     @Override
  58.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause){
  59.         ctx.channel().close();
  60.         clients.remove(ctx.channel());
  61.     }
  64. }

添加监听执行：

Netty 需要在 Spring 加载完成后进行监听执行，因此需要监听 SpringBoot 的 Application

  1. package com.dmbjz;
  3. import com.dmbjz.netty.WsServer;
  4. import org.springframework.context.ApplicationListener;
  5. import org.springframework.context.event.ContextRefreshedEvent;
  6. import org.springframework.stereotype.Component;
  9. /* Spring加载完成后进行监听执行 */
  10. @Component
  11. public class NettyBoot implements ApplicationListener<ContextRefreshedEvent> {
  14.     /* 当一个ApplicationContext被初始化或刷新触发 */
  15.     @Override
  16.     public void onApplicationEvent(ContextRefreshedEvent event) {
  17.         //判断当前是否为spring容器初始化完成，防止重复执行
  18.         if(event.getApplicationContext().getParent() == null){
  19.             try {
  20.                 WsServer.getInstance().start();
  21.             } catch (Exception e) {
  22.                 e.printStackTrace();
  23.             }
  24.         }
  25.     }
  27. }

启动测试：

对之前在 WebSocket 章节中的页面进行修改，使其监听该案例的WebSocket地址，然后进行测试
image.png
访问页面检测监听结果

  1.  ![image.png](https://cdn.nlark.com/yuque/0/2022/png/21405095/1649555166418-2365f7cb-5082-4c7d-954c-02dff6b0973b.png#clientId=u4e14a6b7-98cb-4&crop=0&crop=0&crop=1&crop=1&from=paste&height=310&id=ud773b280&margin=%5Bobject%20Object%5D&name=image.png&originHeight=1394&originWidth=2560&originalType=binary&ratio=1&rotation=0&showTitle=false&size=238988&status=done&style=stroke&taskId=u82cb5f16-d4c6-42db-8281-61c5990167b&title=&width=569)<br />**     输入消息，服务端成功接收**

使用Service：

如果在处理消息时需要使用业务逻辑 Service，只能通过获取 Bean 的方式进行调用，例如在读取状态下需要调用业务方法 UserService ：

  1. import cn.hutool.extra.spring.SpringUtil;
  3. @Override
  4. protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
  6.       UserService userService = SpringUtil.getBean("userServiceImpl");    
  8. }