.Net 4.8中基于websocket-sharp实现WebSocket服务横向扩展方案

针对websocket-sharp在多节点部署下的消息同步问题，核心是解决跨节点的WebSocket消息转发，以下是几种适合.Net 4.8环境的落地方案：

方案1：基于Redis Pub/Sub实现消息中转 #

Redis的发布订阅是轻量高效的跨节点消息同步方案，适合大多数场景：

实现步骤 #

1. 每个WebSocket节点引入StackExchange.Redis NuGet包（选择.Net4.8兼容版本）
2. 每个节点启动时，连接Redis服务器，并订阅指定频道（比如websocket_broadcast）
3. 当节点收到客户端WebSocket消息时：
   • 先处理本地业务逻辑
   • 将消息序列化后发布到Redis频道
4. 其他节点收到Redis频道的消息后，反序列化并推送给本地对应的WebSocket客户端（需维护本地连接映射，比如按用户ID分组）

代码示例 #

// 全局单例Redis连接
private static ConnectionMultiplexer _redis = ConnectionMultiplexer.Connect("your_redis_server:6379");
private static ISubscriber _subscriber = _redis.GetSubscriber();

// WebSocket服务初始化逻辑
public void StartWebSocketServer()
{
    var server = new WebSocketServer("ws://0.0.0.0:8080");
    server.AddWebSocketService<ChatService>("/chat");
    server.Start();

    // 订阅Redis频道，接收跨节点消息
    _subscriber.Subscribe("websocket_broadcast", (channel, message) =>
    {
        var msg = JsonConvert.DeserializeObject<MessageModel>(message);
        ChatService.PushToClients(msg);
    });
}

// 自定义WebSocket处理类
public class ChatService : WebSocketBehavior
{
    // 维护本地连接映射（用户ID到WebSocket连接）
    private static readonly Dictionary<string, ChatService> _connectedClients = new Dictionary<string, ChatService>();

    protected override void OnOpen()
    {
        var userId = Context.QueryString["userId"];
        if (!string.IsNullOrEmpty(userId))
        {
            lock (_connectedClients)
            {
                _connectedClients[userId] = this;
            }
        }
    }

    protected override void OnMessage(MessageEventArgs e)
    {
        // 处理本地业务逻辑
        ProcessLocalMessage(e.Data);

        // 发布到Redis，通知其他节点
        var msg = new MessageModel { Content = e.Data, TargetUserId = GetTargetUserId(e.Data) };
        _subscriber.Publish("websocket_broadcast", JsonConvert.SerializeObject(msg));
    }

    protected override void OnClose(CloseEventArgs e)
    {
        var userId = Context.QueryString["userId"];
        if (!string.IsNullOrEmpty(userId))
        {
            lock (_connectedClients)
            {
                _connectedClients.Remove(userId);
            }
        }
    }

    // 推送给本地匹配的客户端
    public static void PushToClients(MessageModel msg)
    {
        lock (_connectedClients)
        {
            if (_connectedClients.TryGetValue(msg.TargetUserId, out var client))
            {
                client.Send(msg.Content);
            }
        }
    }

    // 解析消息中的目标用户ID
    private string GetTargetUserId(string data)
    {
        // 根据实际消息格式解析，示例逻辑
        var msgObj = JsonConvert.DeserializeObject<dynamic>(data);
        return msgObj.targetUserId;
    }

    // 本地业务处理逻辑
    private void ProcessLocalMessage(string data)
    {
        // 比如记录日志、更新本地缓存等
    }
}

// 消息模型
public class MessageModel
{
    public string Content { get; set; }
    public string TargetUserId { get; set; }
}

方案2：基于RabbitMQ实现可靠消息转发 #

如果需要消息持久化、故障重试等可靠特性，可使用RabbitMQ替代Redis：

实现步骤 #

1. 引入RabbitMQ.Client NuGet包（选择.Net4.8兼容版本）
2. 每个节点作为RabbitMQ的生产者和消费者：
   • 生产者：将本地WebSocket消息发送到RabbitMQ的Exchange
   • 消费者：绑定Exchange到队列，接收消息后推送给本地客户端
3. 使用Fanout Exchange实现广播，或Direct Exchange实现定向推送（按用户ID路由）

核心代码片段 #

// 全局单例RabbitMQ连接
private static IConnection _rabbitConn;
private static IModel _rabbitChannel;

// 初始化RabbitMQ
public void InitRabbitMQ()
{
    var factory = new ConnectionFactory { HostName = "your_rabbitmq_server" };
    _rabbitConn = factory.CreateConnection();
    _rabbitChannel = _rabbitConn.CreateModel();

    // 声明Fanout Exchange
    _rabbitChannel.ExchangeDeclare("websocket_exchange", ExchangeType.Fanout);
    var queueName = _rabbitChannel.QueueDeclare().QueueName;
    _rabbitChannel.QueueBind(queue: queueName, exchange: "websocket_exchange", routingKey: "");

    // 消费消息
    var consumer = new EventingBasicConsumer(_rabbitChannel);
    consumer.Received += (model, ea) =>
    {
        var body = ea.Body.ToArray();
        var message = Encoding.UTF8.GetString(body);
        ChatService.PushToClients(JsonConvert.DeserializeObject<MessageModel>(message));
    };
    _rabbitChannel.BasicConsume(queue: queueName, autoAck: true, consumer: consumer);
}

// 发送消息到RabbitMQ
public void SendToRabbitMQ(MessageModel msg)
{
    var body = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(msg));
    _rabbitChannel.BasicPublish(exchange: "websocket_exchange", routingKey: "", basicProperties: null, body: body);
}

方案3：自定义节点间TCP通信（无第三方依赖） #

如果不想引入中间件，可实现节点间的直接TCP通信：

实现思路 #

1. 每个节点启动时，通过配置文件维护其他节点的TCP地址列表
2. 每个节点启动一个TCP监听服务，接收其他节点的消息
3. 当本地收到WebSocket消息时，遍历所有其他节点的TCP地址，发送消息
4. 其他节点收到TCP消息后，推送给本地客户端

注意事项 #

• 需要自行处理节点故障检测、消息重试逻辑
• 适合节点数量较少的场景，节点过多时维护成本高

关键注意点 #

• 每个节点必须维护本地WebSocket连接池，确保能快速找到目标客户端
• 消息序列化推荐用Newtonsoft.Json，.Net4.8环境成熟稳定
• 生产环境需针对Redis/RabbitMQ做高可用部署，避免单点故障

内容的提问来源于stack exchange，提问作者Kiran