Netty Web Socket Server

WebSocket 的优势

在多次交互场景下，WebSocket 通过以下几种方式显著提升了性能，相比于传统的 HTTP 协议更加适合实时应用场景：

  1. 持久连接:

    • HTTP: 每次请求都需要重新建立和关闭 TCP 连接，这在高频次请求场景下增加了延迟和资源开销。
    • WebSocket: 连接一旦建立后会保持持久，客户端和服务器之间的单个 TCP 连接可以一直保持打开状态，减少了延迟和资源消耗。

  2. 减少开销:

    • HTTP: 每次请求都会附带大量的头信息，增加了数据传输的体积。
    • WebSocket: 仅在连接建立时进行一次 HTTP 握手，之后的数据交换在 WebSocket 协议下进行，减少了每次数据交换的开销。

  3. 全双工通信:

    • HTTP: 只支持半双工通信，通信只能在一个方向上进行。
    • WebSocket: 支持全双工通信，客户端和服务器可以同时发送数据，适合需要实时双向数据交换的场景。

  4. 减少服务器压力:

    • HTTP: 每次请求需要单独的连接，服务器需要处理大量的短连接，增加了 CPU 和内存消耗。
    • WebSocket: 持久连接使服务器只需处理少量的长连接，特别是在大量并发连接的情况下，降低了服务器的开销。

  5. 实时性:

    • HTTP: 常用轮询（Polling）或长轮询（Long Polling）模拟实时性，存在延迟。
    • WebSocket: 服务器可以实时推送数据到客户端，实现无延迟的实时通信。

使用 Netty 开发一个聊天服务器

在这个部分，我们展示了如何使用 Netty 和 tio-boot 框架来开发一个支持 WebSocket 的聊天服务器。以下是核心代码和工作原理的解释。

代码示例

启动类

启动类 ImServerApp 负责初始化和启动整个应用程序。通过 TioApplication.run() 启动应用，AComponentScan 注解则负责扫描和注入相关组件。

import nexus.io.jfinal.aop.annotation.AComponentScan;
import nexus.io.tio.boot.TioApplication;

@AComponentScan
public class ImServerApp {
  public static void main(String[] args) {
    long start = System.currentTimeMillis();
    TioApplication.run(ImServerApp.class, args);
    long end = System.currentTimeMillis();
    System.out.println((end - start) + "ms");
  }
}

Netty 启动配置

NettyServerConfig 类负责配置和启动 Netty 服务器。这里通过 MyChannelInitializer 初始化管道配置，并在指定端口启动 Netty 服务器。

import nexus.io.im.netty.bootstap.NettyServerBootstrap;
import nexus.io.im.netty.handler.MyChannelInitializer;
import nexus.io.jfinal.aop.annotation.AConfiguration;
import nexus.io.jfinal.aop.annotation.AInitialization;
import nexus.io.tio.boot.server.TioBootServer;
import nexus.io.tio.utils.environment.EnvUtils;
import nexus.io.tio.utils.thread.TioThreadUtils;

@AConfiguration
public class NettyServerConfig {

  @Initialization(priority = 101)
  public void nettyServerBootstrap() {
    int nioPort = EnvUtils.getInt("netty.server.port", 7008);

    MyChannelInitializer nettyServerChannelHandler = new MyChannelInitializer();
    NettyServerBootstrap nettyServerBootstrap = new NettyServerBootstrap(nioPort, nettyServerChannelHandler);

    // 启动Netty
    TioThreadUtils.getFixedThreadPool().execute(() -> {
      nettyServerBootstrap.start();
    });

    HookCan.me().addDestroyMethod(nettyServerBootstrap::close);
  }
}

启动 netty

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class NettyServerBootstrap {
  private int port;
  private ChannelHandler nettyServerChannelHandler;
  private EventLoopGroup boss = new NioEventLoopGroup();
  private EventLoopGroup worker = new NioEventLoopGroup();

  public NettyServerBootstrap(int port, ChannelHandler nettyServerChannelHandler) {
    this.port = port;
    this.nettyServerChannelHandler = nettyServerChannelHandler;
  }

  public void start() {
    boss = new NioEventLoopGroup();
    worker = new NioEventLoopGroup();
    try {
      ServerBootstrap bootstrap = new ServerBootstrap();
      bootstrap.group(boss, worker);
      bootstrap.channel(NioServerSocketChannel.class);
      bootstrap.childHandler(nettyServerChannelHandler);
      ChannelFuture f = bootstrap.bind(port).sync();
      if (f.isSuccess()) {
        log.info("Netty started successfully on port: {}", this.port);
      }
      f.channel().closeFuture().sync();
    } catch (Exception e) {
      log.info("Netty start failed: {}", e.getMessage());
      e.printStackTrace();
    } finally {
      close();
    }
  }

  public void close() {
    log.info("Closing Netty");
    if (boss != null) {
      boss.shutdownGracefully();
    }
    if (worker != null) {
      worker.shutdownGracefully();
    }
  }
}

Channel 初始化

MyChannelInitializer 类配置了 Netty 的通道管道，它主要负责添加必要的处理器（Handler），如 HTTP 解码器、WebSocket 协议处理器等。

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
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 MyChannelInitializer extends ChannelInitializer<SocketChannel> {
  @Override
  protected void initChannel(SocketChannel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();

    // WebSocket 基于HTTP协议，所以要有HTTP解码器
    pipeline.addLast(new HttpServerCodec());
    // 对写大数据流的支持
    pipeline.addLast(new ChunkedWriteHandler());
    // 对HTTPMessage进行聚合，聚合成FullHttpRequest或FullHttpResponse
    pipeline.addLast(new HttpObjectAggregator(1024 * 64));
    // WebSocket服务器处理协议
    pipeline.addLast(new WebSocketServerProtocolHandler("/netty"));
    // 自定义的handler
    pipeline.addLast(new ChatHandler());
  }
}

消息处理逻辑

ChatHandler 类负责处理 WebSocket 消息，支持用户注册、消息转发等功能。注册后的用户将被存储在 MessageUtils.nettyUserMap 中，便于消息的转发处理。

import java.util.Map;

import com.alibaba.fastjson2.JSONObject;
import nexus.io.im.netty.utils.MessageUtils;

import io.netty.channel.Channel;
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 lombok.extern.slf4j.Slf4j;

@Slf4j
public class ChatHandler extends SimpleChannelInboundHandler<Object> {
  @Override
  protected void channelRead0(ChannelHandlerContext channelHandlerContext, Object msg) throws Exception {
    if (msg instanceof FullHttpRequest) {
      // 处理 HTTP 请求
    } else if (msg instanceof TextWebSocketFrame) { // 处理 WebSocket 文本消息
      TextWebSocketFrame textWebSocketFrame = (TextWebSocketFrame) msg;
      String content = textWebSocketFrame.text();

      JSONObject msgJson = JSONObject.parseObject(content);
      Integer code = msgJson.getInteger("code");
      if (code == 101) {
        Integer userId = msgJson.getInteger("userId");
        MessageUtils.nettyUserMap.put(userId, channelHandlerContext.channel());
        log.info("netty客户端" + userId + "加入成功！");
        channelHandlerContext.channel().writeAndFlush(new TextWebSocketFrame("User " + userId + " joined successfully!"));
      } else {
        Integer toUserId = msgJson.getInteger("toUserId");
        Channel toUserChannel = MessageUtils.nettyUserMap.get(toUserId);

        if (toUserChannel != null) {
          toUserChannel.writeAndFlush(new TextWebSocketFrame(content));
        } else {
          log.info("用户 " + toUserId + " 未连接。");
        }
      }
    }
  }

  @Override
  public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
    log.info("ctx:{}", ctx);
  }

  @Override
  public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
    Channel channel = ctx.channel();
    Integer userIdToRemove = null;
    for (Map.Entry<Integer, Channel> entry : MessageUtils.nettyUserMap.entrySet()) {
      if (entry.getValue().equals(channel)) {
        userIdToRemove = entry.getKey();
        break;
      }
    }

    if (userIdToRemove != null) {
      MessageUtils.nettyUserMap.remove(userIdToRemove);
      log.info("用户 " + userIdToRemove + " 已断开连接并移除。");
    } else {
      log.info("未找到对应的用户ID。");
    }
  }
}

如何工作

该聊天服务器基于 WebSocket 实现，能够支持客户端之间的实时通信。以下是其工作流程：

  1. 客户端连接和注册：

    • 客户端连接到 WebSocket 服务器后，首先发送包含用户 ID 的注册消息。
    • 服务器接收该消息并将用户的通道注册到 nettyUserMap 中，便于后续的消息转发。

  2. 消息转发：

    • 客户端发送消息给服务器时，消息中会包含接收方的用户 ID。
    • 服务器根据用户 ID 查找对应的通道，并将消息转发给目标用户的通道，完成消息的传递。

  3. 连接管理：

    • 当客户端断开连接时，服务器会从 nettyUserMap 中移除该用户的通道，以确保不再为断开连接的用户转发消息。

通过 Netty 和 tio-boot 的整合，该 WebSocket 服务器能够高效地处理多用户的实时聊天需求，具备了持久连接、全双工通信、低延迟的优势，非常适合需要频繁交互的实时应用场景。

发送消息测试

客户端 1 和客户端 2 能够通过 WebSocket 通信。以下是注册和消息转发的 JSON 消息格式：

1. 客户端注册消息

客户端在连接 WebSocket 服务器后，应该首先发送注册消息，告知服务器其用户 ID。这将允许服务器在 nettyUserMap 中存储用户的通道。

JSON 消息格式：

{
  "code": 101,
  "userId": 1
}

字段解释：

  • "code": 101：标识这是一个注册消息，代码 101 用于区别其他类型的消息。
  • "userId": 1：用户的唯一标识符，在系统中唯一，例如，客户端 1 的用户 ID 可以是 1，客户端 2 的用户 ID 可以是 2。

2. 消息转发

当客户端 1 需要向客户端 2 发送消息时，它应该发送一个包含目标用户 ID 的消息。服务器接收到消息后，会将该消息转发给指定的目标用户。

JSON 消息格式：

{
  "code": 200,
  "fromUserId": 1,
  "toUserId": 2,
  "message": "Hello, this is a message from Client 1 to Client 2."
}

字段解释：

  • "code": 200：标识这是一个需要转发的普通消息。
  • "fromUserId": 1：发送消息的用户 ID（即客户端 1）。
  • "toUserId": 2：接收消息的目标用户 ID（即客户端 2）。
  • "message": "Hello, this is a message from Client 1 to Client 2."：实际发送的消息内容。

3. 示例流程

假设您有两个客户端，客户端 1 和客户端 2：

  1. 客户端 1 注册：

    • 客户端 1 连接到 WebSocket 服务器并发送以下 JSON 消息：
      {
  "code": 101,
  "userId": 1
}
    • 服务器收到后，会将客户端 1 的 Channel 注册到 nettyUserMap 中，键为用户 ID 1。

  2. 客户端 2 注册：

    • 客户端 2 连接到 WebSocket 服务器并发送以下 JSON 消息：
      {
  "code": 101,
  "userId": 2
}
    • 服务器收到后，会将客户端 2 的 Channel 注册到 nettyUserMap 中，键为用户 ID 2。

  3. 客户端 1 向客户端 2 发送消息：

    • 客户端 1 发送一条消息，希望将该消息转发给客户端 2：
      {
  "code": 200,
  "fromUserId": 1,
  "toUserId": 2,
  "message": "Hello, Client 2!"
}
    • 服务器收到消息后，会查找 nettyUserMap 中键为 2 的 Channel，然后将该消息内容通过 WebSocket 转发给客户端 2。

  4. 客户端 2 接收消息：

    • 客户端 2 接收到服务器转发的消息，并处理或显示该消息。

通过这些消息格式，客户端之间可以顺利进行注册和消息转发通信。