feat：封装集群中发送消息方式

Author: Cyaim

Cyaim.WebSocketServer/Cluster/Cyaim.WebSocketServer.Cluster.Hybrid/HybridClusterTransport.cs

@@ -26,7 +26,10 @@ public class HybridClusterTransport : IClusterTransport
         private readonly NodeInfo _nodeInfo;
         private readonly ConcurrentDictionary<string, NodeInfo> _knownNodes;
         private readonly CancellationTokenSource _cancellationTokenSource;
-        private readonly ConcurrentDictionary<string, DateTime> _processedMessageIds; // 已处理的消息ID，用于去重
+        // 已处理的消息ID，用于去重
+        // Key format: "MessageId:FromNodeId" for broadcast messages, "MessageId:FromNodeId:ToNodeId" for targeted messages
+        // 键格式：广播消息为 "MessageId:FromNodeId"，定向消息为 "MessageId:FromNodeId:ToNodeId"
+        private readonly ConcurrentDictionary<string, DateTime> _processedMessageIds;
         private readonly Timer _messageIdCleanupTimer; // 定期清理过期的消息ID
         private bool _disposed = false;
         private bool _started = false;
@@ -217,6 +220,20 @@ public async Task SendAsync(string nodeId, ClusterMessage message)
                 message.FromNodeId = _nodeId;
                 message.ToNodeId = nodeId;
 
+                // Ensure MessageId is set for deduplication / 确保设置 MessageId 以便去重
+                // If MessageId is empty, generate a unique one based on node and timestamp
+                // 如果 MessageId 为空，基于节点和时间戳生成唯一ID
+                if (string.IsNullOrEmpty(message.MessageId))
+                {
+                    message.MessageId = $"{_nodeId}:{nodeId}:{DateTime.UtcNow.Ticks}:{Guid.NewGuid():N}";
+                }
+
+                // Ensure Timestamp is set / 确保设置时间戳
+                if (message.Timestamp == default)
+                {
+                    message.Timestamp = DateTime.UtcNow;
+                }
+
                 var routingKey = $"node.{nodeId}";
                 var messageBytes = Encoding.UTF8.GetBytes(JsonSerializer.Serialize(message));
 
@@ -254,6 +271,20 @@ public async Task BroadcastAsync(ClusterMessage message)
                 message.FromNodeId = _nodeId;
                 message.ToNodeId = null; // Broadcast / 广播
 
+                // Ensure MessageId is set for deduplication / 确保设置 MessageId 以便去重
+                // If MessageId is empty, generate a unique one
+                // 如果 MessageId 为空，生成唯一ID
+                if (string.IsNullOrEmpty(message.MessageId))
+                {
+                    message.MessageId = $"{_nodeId}:broadcast:{DateTime.UtcNow.Ticks}:{Guid.NewGuid():N}";
+                }
+
+                // Ensure Timestamp is set / 确保设置时间戳
+                if (message.Timestamp == default)
+                {
+                    message.Timestamp = DateTime.UtcNow;
+                }
+
                 var messageBytes = Encoding.UTF8.GetBytes(JsonSerializer.Serialize(message));
 
                 var properties = new MessageProperties
@@ -274,6 +305,73 @@ public async Task BroadcastAsync(ClusterMessage message)
             }
         }
 
+        /// <summary>
+        /// Send message to multiple nodes / 向多个节点发送消息
+        /// </summary>
+        /// <param name="nodeIds">Target node IDs / 目标节点 ID 列表</param>
+        /// <param name="message">Message to send / 要发送的消息</param>
+        /// <returns>Dictionary of node ID and send result / 节点ID和发送结果的字典</returns>
+        /// <remarks>
+        /// This method sends the same message to multiple nodes. Each node will receive a copy of the message.
+        /// If you want to send different messages to different nodes, call SendAsync multiple times.
+        /// 此方法向多个节点发送相同的消息。每个节点都会收到消息的副本。
+        /// 如果要向不同节点发送不同的消息，请多次调用 SendAsync。
+        /// </remarks>
+        public async Task<Dictionary<string, bool>> SendToMultipleNodesAsync(IEnumerable<string> nodeIds, ClusterMessage message)
+        {
+            if (nodeIds == null)
+            {
+                throw new ArgumentNullException(nameof(nodeIds));
+            }
+
+            if (message == null)
+            {
+                throw new ArgumentNullException(nameof(message));
+            }
+
+            var results = new Dictionary<string, bool>();
+            var tasks = new List<Task>();
+
+            foreach (var nodeId in nodeIds)
+            {
+                if (string.IsNullOrEmpty(nodeId))
+                {
+                    continue;
+                }
+
+                var task = Task.Run(async () =>
+                {
+                    try
+                    {
+                        // Create a copy of the message for each node to ensure unique MessageId per node
+                        // 为每个节点创建消息副本，确保每个节点都有唯一的 MessageId
+                        var nodeMessage = new ClusterMessage
+                        {
+                            Type = message.Type,
+                            FromNodeId = message.FromNodeId,
+                            ToNodeId = nodeId,
+                            Payload = message.Payload,
+                            MessageId = message.MessageId, // Will be auto-generated if empty / 如果为空将自动生成
+                            Timestamp = message.Timestamp
+                        };
+
+                        await SendAsync(nodeId, nodeMessage);
+                        results[nodeId] = true;
+                    }
+                    catch (Exception ex)
+                    {
+                        _logger.LogError(ex, $"Failed to send message to node {nodeId}");
+                        results[nodeId] = false;
+                    }
+                });
+
+                tasks.Add(task);
+            }
+
+            await Task.WhenAll(tasks);
+            return results;
+        }
+
         /// <summary>
         /// Check if node is connected / 检查节点是否已连接
         /// </summary>
@@ -373,23 +471,44 @@ private async Task<bool> HandleMessageAsync(byte[] body, MessageProperties prope
                 if (!string.IsNullOrEmpty(message.ToNodeId) && message.ToNodeId != _nodeId)
                 {
                     // This message is for another node, ignore it / 此消息是发送给其他节点的，忽略它
-                    _logger.LogDebug($"Ignoring message {message.MessageId} - target node is {message.ToNodeId}, current node is {_nodeId}");
+                    // Note: This is expected behavior for targeted messages. Use BroadcastAsync to send to all nodes.
+                    // 注意：这是定向消息的预期行为。使用 BroadcastAsync 可以向所有节点发送消息。
+                    _logger.LogTrace($"Ignoring message {message.MessageId} (type: {message.Type}) - target node is {message.ToNodeId}, current node is {_nodeId}. This is normal for targeted messages.");
                     return true; // Ack to remove from queue / 确认以从队列中移除
                 }
 
                 // Check for duplicate messages using message ID / 使用消息ID检查重复消息
+                // For broadcast messages, use "MessageId:FromNodeId" as key to allow same message from same sender to be processed once per node
+                // For targeted messages, use "MessageId:FromNodeId:ToNodeId" to allow same message to different targets
+                // 对于广播消息，使用 "MessageId:FromNodeId" 作为键，允许来自同一发送者的相同消息在每个节点上处理一次
+                // 对于定向消息，使用 "MessageId:FromNodeId:ToNodeId" 允许相同消息发送到不同目标
                 if (!string.IsNullOrEmpty(message.MessageId))
                 {
+                    // Create unique key based on message type / 根据消息类型创建唯一键
+                    string deduplicationKey;
+                    if (string.IsNullOrEmpty(message.ToNodeId))
+                    {
+                        // Broadcast message: same message from same sender should be processed once per node
+                        // 广播消息：来自同一发送者的相同消息应在每个节点上处理一次
+                        deduplicationKey = $"{message.MessageId}:{message.FromNodeId}";
+                    }
+                    else
+                    {
+                        // Targeted message: same message can be sent to different nodes
+                        // 定向消息：相同消息可以发送到不同节点
+                        deduplicationKey = $"{message.MessageId}:{message.FromNodeId}:{message.ToNodeId}";
+                    }
+
                     // Check if we've already processed this message / 检查是否已处理过此消息
-                    if (_processedMessageIds.TryGetValue(message.MessageId, out var processedTime))
+                    if (_processedMessageIds.TryGetValue(deduplicationKey, out var processedTime))
                     {
                         // Message already processed, ignore it / 消息已处理，忽略它
-                        _logger.LogDebug($"Ignoring duplicate message {message.MessageId} (processed at {processedTime:yyyy-MM-dd HH:mm:ss})");
+                        _logger.LogDebug($"Ignoring duplicate message {message.MessageId} (key: {deduplicationKey}, processed at {processedTime:yyyy-MM-dd HH:mm:ss})");
                         return true; // Ack to remove from queue / 确认以从队列中移除
                     }
 
                     // Mark message as processed / 标记消息为已处理
-                    _processedMessageIds.TryAdd(message.MessageId, DateTime.UtcNow);
+                    _processedMessageIds.TryAdd(deduplicationKey, DateTime.UtcNow);
                 }
 
                 // Trigger message received event / 触发消息接收事件

Cyaim.WebSocketServer/Cyaim.WebSocketServer.MessagePack/MessagePackExtensions.cs

@@ -1,15 +1,18 @@
 using System;
+using System.Collections.Generic;
 using System.Net.WebSockets;
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
 using Cyaim.WebSocketServer.Infrastructure;
+using Cyaim.WebSocketServer.Infrastructure.Cluster;
 using MessagePack;
 
 namespace Cyaim.WebSocketServer.MessagePack
 {
     /// <summary>
     /// MessagePack serialization extensions for WebSocket
+    /// MessagePack 序列化扩展方法
     /// </summary>
     public static class MessagePackExtensions
     {
@@ -41,7 +44,7 @@ public static async Task SendAsync<T>(
 
             options ??= MessagePackSerializerOptions.Standard;
             var serializedData = MessagePackSerializer.Serialize(data, options);
-            await WebSocketManager.SendAsync(serializedData, messageType, serializedData.Length <= sendBufferSize, cancellationToken: cancellationToken ?? CancellationToken.None, timeout, sendBufferSize: (uint)sendBufferSize, sockets: socket);
+            await WebSocketManager.SendLocalAsync(serializedData, messageType, serializedData.Length <= sendBufferSize, cancellationToken: cancellationToken ?? CancellationToken.None, timeout, sendBufferSize: (uint)sendBufferSize, sockets: socket);
         }
 
         /// <summary>
@@ -72,8 +75,74 @@ public static async Task SendAsync<T>(
 
             options ??= MessagePackSerializerOptions.Standard;
             var serializedData = MessagePackSerializer.Serialize(data, options);
-            await WebSocketManager.SendAsync(serializedData, messageType, serializedData.Length <= sendBufferSize, cancellationToken: cancellationToken ?? CancellationToken.None, timeout, sendBufferSize: (uint)sendBufferSize, sockets: socket);
+            await WebSocketManager.SendLocalAsync(serializedData, messageType, serializedData.Length <= sendBufferSize, cancellationToken: cancellationToken ?? CancellationToken.None, timeout, sendBufferSize: (uint)sendBufferSize, sockets: socket);
         }
+
+        #region ClusterManager Extensions / 集群管理器扩展
+
+        /// <summary>
+        /// Route object as MessagePack binary message to a connection (supports cross-node)
+        /// 将对象序列化为 MessagePack 二进制消息路由到连接（支持跨节点）
+        /// </summary>
+        /// <typeparam name="T">Object type / 对象类型</typeparam>
+        /// <param name="clusterManager">Cluster manager instance / 集群管理器实例</param>
+        /// <param name="connectionId">Connection ID / 连接 ID</param>
+        /// <param name="data">Object to serialize / 要序列化的对象</param>
+        /// <param name="options">MessagePack serializer options / MessagePack 序列化选项</param>
+        /// <returns>True if routed successfully / 路由成功返回 true</returns>
+        public static async Task<bool> RouteMessagePackAsync<T>(
+            this ClusterManager clusterManager,
+            string connectionId,
+            T data,
+            MessagePackSerializerOptions options = null)
+        {
+            if (clusterManager == null)
+            {
+                throw new ArgumentNullException(nameof(clusterManager));
+            }
+
+            if (data == null)
+            {
+                return false;
+            }
+
+            options ??= MessagePackSerializerOptions.Standard;
+            var bytes = MessagePackSerializer.Serialize(data, options);
+            return await clusterManager.RouteMessageAsync(connectionId, bytes, (int)WebSocketMessageType.Binary);
+        }
+
+        /// <summary>
+        /// Route object as MessagePack binary message to multiple connections (supports cross-node)
+        /// 将对象序列化为 MessagePack 二进制消息路由到多个连接（支持跨节点）
+        /// </summary>
+        /// <typeparam name="T">Object type / 对象类型</typeparam>
+        /// <param name="clusterManager">Cluster manager instance / 集群管理器实例</param>
+        /// <param name="connectionIds">Connection IDs / 连接 ID 列表</param>
+        /// <param name="data">Object to serialize / 要序列化的对象</param>
+        /// <param name="options">MessagePack serializer options / MessagePack 序列化选项</param>
+        /// <returns>Dictionary of connection ID to routing result / 连接ID到路由结果的字典</returns>
+        public static async Task<Dictionary<string, bool>> RouteMessagePacksAsync<T>(
+            this ClusterManager clusterManager,
+            IEnumerable<string> connectionIds,
+            T data,
+            MessagePackSerializerOptions options = null)
+        {
+            if (clusterManager == null)
+            {
+                throw new ArgumentNullException(nameof(clusterManager));
+            }
+
+            if (data == null)
+            {
+                return new Dictionary<string, bool>();
+            }
+
+            options ??= MessagePackSerializerOptions.Standard;
+            var bytes = MessagePackSerializer.Serialize(data, options);
+            return await clusterManager.RouteMessagesAsync(connectionIds, bytes, (int)WebSocketMessageType.Binary);
+        }
+
+        #endregion
     }
 }
 

Cyaim.WebSocketServer/Cyaim.WebSocketServer.MessagePack/README.md

@@ -149,11 +149,64 @@ public class MessagePackResponseScheme
 }
 ```
 
+## 集群支持
+
+### 使用 MessagePack 发送消息到集群
+
+MessagePack 扩展提供了便捷的方法，可以直接通过集群管理器发送 MessagePack 序列化的对象：
+
+```csharp
+using Cyaim.WebSocketServer.MessagePack;
+using Cyaim.WebSocketServer.Infrastructure.Cluster;
+
+// 获取集群管理器
+var clusterManager = GlobalClusterCenter.ClusterManager;
+
+// 定义要发送的对象
+var user = new { Id = 1, Name = "Alice", Age = 30 };
+
+// 向单个连接发送 MessagePack 序列化的对象（支持跨节点）
+await clusterManager.RouteMessagePackAsync("connection-1", user);
+
+// 向多个连接批量发送 MessagePack 序列化的对象（支持跨节点）
+var connectionIds = new[] { "connection-1", "connection-2", "connection-3" };
+var results = await clusterManager.RouteMessagePacksAsync(connectionIds, user);
+
+// 检查发送结果
+foreach (var result in results)
+{
+    if (result.Value)
+    {
+        Console.WriteLine($"成功发送到连接 {result.Key}");
+    }
+    else
+    {
+        Console.WriteLine($"发送到连接 {result.Key} 失败");
+    }
+}
+
+// 使用自定义 MessagePack 选项
+var options = MessagePackSerializerOptions.Standard.WithCompression(MessagePackCompression.Lz4Block);
+await clusterManager.RouteMessagePackAsync("connection-1", user, options);
+```
+
+### 扩展方法说明
+
+- **`RouteMessagePackAsync<T>`**: 向单个连接发送 MessagePack 序列化的对象
+- **`RouteMessagePacksAsync<T>`**: 向多个连接批量发送 MessagePack 序列化的对象
+
+这些方法会自动：
+
+- 将对象序列化为 MessagePack 二进制格式
+- 通过集群路由系统发送到目标连接（支持跨节点）
+- 使用 `WebSocketMessageType.Binary` 消息类型
+
 ## 注意事项
 
 1. **消息类型** - MessagePackChannelHandler 只处理二进制消息 (`WebSocketMessageType.Binary`)
 2. **兼容性** - 内部仍使用 JSON 进行参数绑定，以保持与现有端点的兼容性
 3. **客户端** - 需要支持 MessagePack 的客户端库
+4. **集群扩展** - 使用集群扩展方法需要引用 `Cyaim.WebSocketServer.Infrastructure.Cluster` 命名空间
 
 ## 许可证