使用 tio-boot 搭建 WebSocket 服务

import nexus.io.annotation.AConfiguration;
import nexus.io.annotation.Initialization;
import nexus.io.tio.boot.server.TioBootServer;
import nexus.io.tio.boot.websocket.WebSocketRouter;

@AConfiguration
public class WebSocketConfig {

  @Initialization
  public void config() {
    WebSocketRouter router = TioBootServer.me().getWebSocketRouter();
    router.add("/hello", new HelloWebSocketHandler());
  }
}

package nexus.io.websockify.handler;

import nexus.io.tio.core.ChannelContext;
import nexus.io.tio.core.Tio;
import nexus.io.tio.http.common.HttpRequest;
import nexus.io.tio.http.common.HttpResponse;
import nexus.io.tio.websocket.common.WebSocketRequest;
import nexus.io.tio.websocket.common.WebSocketResponse;
import nexus.io.tio.websocket.common.WebSocketSessionContext;
import nexus.io.tio.websocket.server.handler.IWebSocketHandler;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class HelloWebSocketHandler implements IWebSocketHandler {
  public static final String CHARSET = "utf-8";

  /**
   * 握手成功后执行，绑定群组并通知其他用户
   */
  @Override
  public HttpResponse handshake(HttpRequest httpRequest, HttpResponse response, ChannelContext channelContext) throws Exception {
    log.info("请求信息: {}", httpRequest);
    return response;
  }

  /**
   * 握手完成后执行
   */
  @Override
  public void onAfterHandshaked(HttpRequest httpRequest, HttpResponse httpResponse, ChannelContext channelContext) throws Exception {
    log.info("握手完成: {}", httpRequest);
  }

  /**
   * 处理连接关闭请求，进行资源清理
   */
  @Override
  public Object onClose(WebSocketRequest wsRequest, byte[] bytes, ChannelContext channelContext) throws Exception {
    Tio.remove(channelContext, "客户端主动关闭连接");
    return null;
  }

  /**
   * 处理二进制消息
   */
  @Override
  public Object onBytes(WebSocketRequest wsRequest, byte[] bytes, ChannelContext channelContext) throws Exception {
    log.info("收到二进制消息，大小: {} bytes", bytes.length);
    return null;
  }

  /**
   * 处理文本消息
   */
  @Override
  public Object onText(WebSocketRequest wsRequest, String text, ChannelContext channelContext) throws Exception {
    WebSocketSessionContext wsSessionContext = (WebSocketSessionContext) channelContext.get();
    String path = wsSessionContext.getHandshakeRequest().getRequestLine().path;
    log.info("路径：{}，收到消息：{}", path, text);

    String message = "{user_id:'" + channelContext.userId + "',message:'" + text + "'}";
    WebSocketResponse wsResponse = WebSocketResponse.fromText(message, CHARSET);
    // 发送消息
    Tio.send(channelContext, wsResponse);
    return null; // 不需要额外的返回值
  }
}

package nexus.io.websocket.handler;

import java.util.Objects;

import nexus.io.tio.core.ChannelContext;
import nexus.io.tio.core.Tio;
import nexus.io.tio.http.common.HttpRequest;
import nexus.io.tio.http.common.HttpResponse;
import nexus.io.tio.websocket.common.WebSocketRequest;
import nexus.io.tio.websocket.common.WebSocketResponse;
import nexus.io.tio.websocket.common.WebSocketSessionContext;
import nexus.io.tio.websocket.server.handler.IWebSocketHandler;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class HelloWebSocketHandler implements IWebSocketHandler {

  /**
   * 群组ID，用于群聊
   */
  public static final String GROUP_ID = "group-01";
  public static final String CHARSET = "utf-8";

  /**
   * 握手成功后执行，绑定群组并通知其他用户
   */
  @Override
  public HttpResponse handshake(HttpRequest request, HttpResponse response, ChannelContext channelContext) throws Exception {
    String clientIp = request.getClientIp();
    String name = request.getParam("name");

    Tio.bindUser(channelContext, name);
    log.info("收到来自 {} 的 WebSocket 握手请求：{}", clientIp, request.toString());
    return response;
  }

  /**
   * 握手完成后执行
   */
  @Override
  public void onAfterHandshaked(HttpRequest httpRequest, HttpResponse httpResponse, ChannelContext channelContext) throws Exception {
    // 绑定到群组，便于消息广播
    Tio.bindGroup(channelContext, GROUP_ID);
    // 获取当前在线用户数量
    int count = Tio.getAll(channelContext.tioConfig).getObj().size();

    String message = "{name:'admin',message:'" + channelContext.userid + " 进入聊天室，当前在线人数：" + count + "'}";
    WebSocketResponse wsResponse = WebSocketResponse.fromText(message, CHARSET);
    // 广播消息给群组内所有用户
    Tio.sendToGroup(channelContext.tioConfig, GROUP_ID, wsResponse);
  }

  /**
   * 处理二进制消息
   */
  @Override
  public Object onBytes(WebSocketRequest wsRequest, byte[] bytes, ChannelContext channelContext) throws Exception {
    // 当前示例未处理二进制消息，移除连接
    Tio.remove(channelContext, "客户端发送二进制消息，连接关闭");
    return null;
  }

  /**
   * 处理连接关闭请求，进行资源清理
   */
  @Override
  public Object onClose(WebSocketRequest wsRequest, byte[] bytes, ChannelContext channelContext) throws Exception {
    // 资源清理由 Tio 自动处理，此处可添加额外逻辑
    log.info("连接关闭，用户：{}", channelContext.userid);
    return null;
  }

  /**
   * 处理文本消息
   */
  @Override
  public Object onText(WebSocketRequest wsRequest, String text, ChannelContext channelContext) throws Exception {
    WebSocketSessionContext wsSessionContext = (WebSocketSessionContext) channelContext.get();
    String path = wsSessionContext.getHandshakeRequest().getRequestLine().path;
    log.info("路径：{}，收到消息：{}", path, text);

    if ("心跳内容".equals(text)) {
      return null; // 忽略心跳消息
    }

    String message = "{name:'" + channelContext.userid + "',message:'" + text + "'}";
    WebSocketResponse wsResponse = WebSocketResponse.fromText(message, CHARSET);
    // 广播消息给群组内所有用户
    Tio.sendToGroup(channelContext.tioConfig, GROUP_ID, wsResponse);

    return null; // 不需要额外的返回值
  }
}

import nexus.io.tio.core.ChannelContext;
import nexus.io.tio.http.common.HttpRequest;
import nexus.io.tio.http.common.HttpResponse;
import nexus.io.tio.websocket.common.WebSocketRequest;

public interface IWebSocketHandler {
  /**
   * 对 httpResponse 参数进行补充并返回，如果返回 null 表示不想与对方建立连接，框架会断开连接。如果返回非 null，框架会将这个对象发送给对方。
   * 注意：请不要在这个方法中向对方发送任何消息，因为握手还未完成，发送消息会导致协议交互失败。
   * 对于大部分业务，该方法只需要一行代码：return httpResponse;
   *
   * @param httpRequest 客户端的握手请求
   * @param httpResponse 服务器的响应对象
   * @param channelContext 当前连接的上下文信息
   * @return 修改后的 HttpResponse 对象，或 null 以拒绝握手
   * @throws Exception
   */
  HttpResponse handshake(HttpRequest httpRequest, HttpResponse httpResponse, ChannelContext channelContext) throws Exception;

  /**
   * 握手成功后触发该方法。
   *
   * @param httpRequest 客户端的握手请求
   * @param httpResponse 服务器的响应对象
   * @param channelContext 当前连接的上下文信息
   * @throws Exception
   */
  void onAfterHandshaked(HttpRequest httpRequest, HttpResponse httpResponse, ChannelContext channelContext) throws Exception;

  /**
   * 当收到 Opcode.BINARY 消息时执行该方法。也就是说，如果的 WebSocket 是基于 BINARY 传输的，就会调用此方法。
   *
   * @param wsRequest WebSocket 请求对象
   * @param bytes 消息的二进制内容
   * @param channelContext 当前连接的上下文信息
   * @return 可以是 WebSocketResponse、byte[]、ByteBuffer、String 或 null。如果是 null，框架不会回消息
   * @throws Exception
   */
  Object onBytes(WebSocketRequest wsRequest, byte[] bytes, ChannelContext channelContext) throws Exception;

  /**
   * 当收到 Opcode.CLOSE 时执行该方法，业务层在该方法中一般不需要写逻辑，可以为空。
   *
   * @param wsRequest WebSocket 请求对象
   * @param bytes 消息的二进制内容
   * @param channelContext 当前连接的上下文信息
   * @return 可以是 WebSocketResponse、byte[]、ByteBuffer、String 或 null。如果是 null，框架不会回消息
   * @throws Exception
   */
  Object onClose(WebSocketRequest wsRequest, byte[] bytes, ChannelContext channelContext) throws Exception;

  /**
   * 当收到 Opcode.TEXT 消息时执行该方法。也就是说，如果的 WebSocket 是基于 TEXT 传输的，就会调用此方法。
   *
   * @param wsRequest WebSocket 请求对象
   * @param text 消息的文本内容
   * @param channelContext 当前连接的上下文信息
   * @return 可以是 WebSocketResponse、byte[]、ByteBuffer、String 或 null。如果是 null，框架不会回消息
   * @throws Exception
   */
  Object onText(WebSocketRequest wsRequest, String text, ChannelContext channelContext) throws Exception;
}