添加MAVEN依赖

<!-- Netty -->
<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.45.Final</version>
</dependency>

WebSocketProperties配置类

为了能在SpringBoot项目中灵活配置相关的值，这里使用了配置类，并使用了默认值。

@Getter
@Setter
@Component
@ConfigurationProperties(prefix = "chat.websocket")
public class WebSocketProperties {

  private String host = "localhost"; // 监听地址
  private Integer port = 8081; // 监听端口
  private String path = "/ws"; // 请求路径

}

WebSocket连接通道池

用来管理已经连接的客户端通道，方便数据传输交互。

@Slf4j
@Component
public class NioWebSocketChannelPool {

  private final ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

  /**
   * 新增一个客户端通道
   *
   * @param channel
   */
  public void addChannel(Channel channel) {
    channels.add(channel);
  }

  /**
   * 移除一个客户端连接通道
   *
   * @param channel
   */
  public void removeChannel(Channel channel) {
    channels.remove(channel);
  }

}

WebSocket连接数据接收处理类

@Slf4j
@Sharable
@Component
public class NioWebSocketHandler extends SimpleChannelInboundHandler<WebSocketFrame> {

  @Autowired
  private NioWebSocketChannelPool webSocketChannelPool;
  @Autowired
  private WebSocketProperties webSocketProperties;

  @Override
  public void channelActive(ChannelHandlerContext ctx) throws Exception {
    log.debug("客户端连接：{}", ctx.channel().id());
    webSocketChannelPool.addChannel(ctx.channel());
    super.channelActive(ctx);
  }

  @Override
  public void channelInactive(ChannelHandlerContext ctx) throws Exception {
    log.debug("客户端断开连接：{}", ctx.channel().id());
    webSocketChannelPool.removeChannel(ctx.channel());
    super.channelInactive(ctx);
  }

  @Override
  public void channelReadComplete(ChannelHandlerContext ctx) {
    ctx.channel().flush();
  }

  @Override
  protected void channelRead0(ChannelHandlerContext ctx, WebSocketFrame frame) {
    if (frame instanceof PingWebSocketFrame) {
      pingWebSocketFrameHandler(ctx, (PingWebSocketFrame) frame);
    } else if (frame instanceof TextWebSocketFrame) {
      textWebSocketFrameHandler(ctx, (TextWebSocketFrame) frame);
    } else if (frame instanceof CloseWebSocketFrame) {
      closeWebSocketFrameHandler(ctx, (CloseWebSocketFrame) frame);
    }
  }

  @Override
  public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    log.info("数据类型：{}", msg.getClass());
    if (msg instanceof FullHttpRequest) {
      fullHttpRequestHandler(ctx, (FullHttpRequest) msg);
    }
    super.channelRead(ctx, msg);
  }

  /**
   * 处理连接请求，客户端WebSocket发送握手包时会执行这一次请求
   *
   * @param ctx
   * @param request
   */
  private void fullHttpRequestHandler(ChannelHandlerContext ctx, FullHttpRequest request) {
    String uri = request.uri();
    log.debug("接收到客户端的握手包：{}", ctx.channel().id());
    Map<String, String> params = RequestUriUtils.getParams(uri);
    log.debug("客户端请求参数：{}", params);
    if (uri.startsWith(webSocketProperties.getPath()))
      request.setUri(webSocketProperties.getPath());
    else
      ctx.close();
  }

  /**
   * 客户端发送断开请求处理
   *
   * @param ctx
   * @param frame
   */
  private void closeWebSocketFrameHandler(ChannelHandlerContext ctx, CloseWebSocketFrame frame) {
    log.debug("接收到主动断开请求：{}", ctx.channel().id());
    ctx.close();
  }

  /**
   * 创建连接之后，客户端发送的消息都会在这里处理
   *
   * @param ctx
   * @param frame
   */
  private void textWebSocketFrameHandler(ChannelHandlerContext ctx, TextWebSocketFrame frame) {
    String text = frame.text();
    log.debug("接收到客户端的消息：{}", text);
    // 将客户端消息回送给客户端
    ctx.channel().writeAndFlush(new TextWebSocketFrame("你发送的内容是：" + text));
  }

  /**
   * 处理客户端心跳包
   *
   * @param ctx
   * @param frame
   */
  private void pingWebSocketFrameHandler(ChannelHandlerContext ctx, PingWebSocketFrame frame) {
    ctx.channel().writeAndFlush(new PongWebSocketFrame(frame.content().retain()));
  }
}

WebSocket通道连接初始化

@Component
public class NioWebSocketChannelInitializer extends ChannelInitializer<SocketChannel> {

  @Autowired
  private WebSocketProperties webSocketProperties;
  @Autowired
  private NioWebSocketHandler nioWebSocketHandler;

  @Override
  protected void initChannel(SocketChannel socketChannel) {
    socketChannel.pipeline()
        .addLast(new HttpServerCodec())
        .addLast(new ChunkedWriteHandler())
        .addLast(new HttpObjectAggregator(8192))
        .addLast(nioWebSocketHandler)
        .addLast(new WebSocketServerProtocolHandler(webSocketProperties.getPath(), null, true, 65536));
  }
}

Netty服务端

服务器的初始化和销毁都交给Spring容器。

@Slf4j
@Component
public class NioWebSocketServer implements InitializingBean, DisposableBean {

  @Autowired
  private WebSocketProperties webSocketProperties;
  @Autowired
  private NioWebSocketChannelInitializer webSocketChannelInitializer;

  private EventLoopGroup bossGroup;
  private EventLoopGroup workGroup;
  private ChannelFuture channelFuture;

  @Override
  public void afterPropertiesSet() throws Exception {
    try {
      bossGroup = new NioEventLoopGroup(webSocketProperties.getBoss());
      workGroup = new NioEventLoopGroup(webSocketProperties.getWork());

      ServerBootstrap serverBootstrap = new ServerBootstrap();
      serverBootstrap.option(ChannelOption.SO_BACKLOG, 1024)
          .group(bossGroup, workGroup)
          .channel(NioServerSocketChannel.class)
          .localAddress(webSocketProperties.getPort())
          .childHandler(webSocketChannelInitializer);

      channelFuture = serverBootstrap.bind().sync();
    } finally {
      if (channelFuture != null && channelFuture.isSuccess()) {
        log.info("Netty server startup on port: {} (websocket) with context path '{}'", webSocketProperties.getPort(), webSocketProperties.getPath());
      } else {
        log.error("Netty server startup failed.");
        if (bossGroup != null)
          bossGroup.shutdownGracefully().sync();
        if (workGroup != null)
          workGroup.shutdownGracefully().sync();
      }
    }
  }

  @Override
  public void destroy() throws Exception {
    log.info("Shutting down Netty server...");
    if (bossGroup != null && !bossGroup.isShuttingDown() && !bossGroup.isTerminated())
      bossGroup.shutdownGracefully().sync();
    if (workGroup != null && !workGroup.isShuttingDown() && !workGroup.isTerminated())
      workGroup.shutdownGracefully().sync();
    if (channelFuture != null && channelFuture.isSuccess())
      channelFuture.channel().closeFuture().sync();
    log.info("Netty server shutdown.");
  }
}

到此，代码编写完成了。