Skip to content

Commit 763a168

Browse files
CyaimCyaim
committed
merge
2 parents a0e2945 + 3285438 commit 763a168

7 files changed

Lines changed: 443 additions & 66 deletions

File tree

‎Cyaim.WebSocketServer/Cluster/Cyaim.WebSocketServer.Cluster.Hybrid.MessageQueue.RabbitMQ/Cyaim.WebSocketServer.Cluster.Hybrid.MessageQueue.RabbitMQ.csproj‎

Lines changed: 4 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
<?xml version="1.0" encoding="UTF-8"?>
1+
<?xml version="1.0" encoding="utf-8"?>
22
<Project Sdk="Microsoft.NET.Sdk">
33
<PropertyGroup>
44
<TargetFrameworks>netstandard2.1;net6.0;net7.0;net8.0;net9.0;net10.0</TargetFrameworks>
@@ -29,16 +29,12 @@
2929
</ItemGroup>
3030
<!-- .NET Standard 2.1 需要显式引用，使用最新稳定版�? -->
3131
<ItemGroup Condition="'$(TargetFramework)' == 'netstandard2.1'">
32-
<PackageReference Include="Microsoft.Extensions.Logging.Abstractions"
33-
Version="9.0.0" />
34-
<PackageReference Include="System.Text.Json"
35-
Version="9.0.0" />
32+
<PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="9.0.0" />
33+
<PackageReference Include="System.Text.Json" Version="9.0.0" />
3634
</ItemGroup>
3735
<!-- RabbitMQ.Client 7.0+ 支持 .NET Standard 2.0+，统一使用 7.2.0 -->
3836
<ItemGroup>
39-
<PackageReference Include="RabbitMQ.Client"
40-
Version="7.2.0" />
41-
<ProjectReference Include="..\Cyaim.WebSocketServer.Cluster.Hybrid\Cyaim.WebSocketServer.Cluster.Hybrid.csproj" />
37+
<PackageReference Include="RabbitMQ.Client" Version="7.2.0" />
4238
</ItemGroup>
4339
<ItemGroup>
4440
<None Include="..\..\..\LICENSE">

‎Cyaim.WebSocketServer/Cluster/Cyaim.WebSocketServer.Cluster.Hybrid.MessageQueue.RabbitMQ/RabbitMQMessageQueueService.cs‎

Lines changed: 337 additions & 4 deletions
Large diffs are not rendered by default.

‎Cyaim.WebSocketServer/Cluster/Cyaim.WebSocketServer.Cluster.Hybrid.Redis.FreeRedis/Cyaim.WebSocketServer.Cluster.Hybrid.Redis.FreeRedis.csproj‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,6 @@
3636

3737
<ItemGroup>
3838
<PackageReference Include="FreeRedis" Version="1.2.0" />
39-
<ProjectReference Include="..\Cyaim.WebSocketServer.Cluster.Hybrid\Cyaim.WebSocketServer.Cluster.Hybrid.csproj" />
4039
</ItemGroup>
4140

4241
<ItemGroup>

‎Cyaim.WebSocketServer/Cluster/Cyaim.WebSocketServer.Cluster.Hybrid/Cyaim.WebSocketServer.Cluster.Hybrid.csproj‎

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -39,10 +39,6 @@
3939

4040
<!-- .NET 6.0+ 框架已内置这些包，无需显式引用，框架会自动使用内置版本 -->
4141

42-
<ItemGroup>
43-
<ProjectReference Include="..\..\Cyaim.WebSocketServer\Cyaim.WebSocketServer.csproj" />
44-
</ItemGroup>
45-
4642
<ItemGroup>
4743
<None Include="..\..\..\LICENSE">
4844
<Pack>True</Pack>

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

Lines changed: 65 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -136,6 +136,30 @@ public async Task StartAsync()
136136

137137
// Connect to message queue / 连接到消息队列
138138
await _messageQueueService.ConnectAsync();
139+
140+
// Verify connection is ready before proceeding / 验证连接已就绪再继续
141+
// This ensures channel is fully established before RaftNode starts / 这确保在 RaftNode 启动前 channel 已完全建立
142+
int verifyAttempts = 0;
143+
while (verifyAttempts < 10) // Try up to 10 times / 最多尝试 10 次
144+
{
145+
try
146+
{
147+
// Try to ensure connection is ready / 尝试确保连接已就绪
148+
await _messageQueueService.VerifyConnectionAsync();
149+
break; // Connection verified / 连接已验证
150+
}
151+
catch (Exception ex)
152+
{
153+
verifyAttempts++;
154+
if (verifyAttempts >= 10)
155+
{
156+
_logger.LogError(ex, $"[HybridClusterTransport] 验证 RabbitMQ 连接失败，已达到最大重试次数 - NodeId: {_nodeId}");
157+
throw;
158+
}
159+
_logger.LogWarning($"[HybridClusterTransport] 验证 RabbitMQ 连接失败，重试中 ({verifyAttempts}/10) - NodeId: {_nodeId}, Error: {ex.Message}");
160+
await Task.Delay(200); // Wait 200ms before retry / 重试前等待 200ms
161+
}
162+
}
139163

140164
// Declare exchange / 声明交换机
141165
await _messageQueueService.DeclareExchangeAsync(ExchangeName, "topic", durable: true);
@@ -154,7 +178,8 @@ public async Task StartAsync()
154178
await _messageQueueService.BindQueueAsync(_queueName, ExchangeName, BroadcastRoutingKey);
155179

156180
// Start consuming messages / 开始消费消息
157-
await _messageQueueService.ConsumeAsync(_queueName, HandleMessageAsync, autoAck: false);
181+
// Pass currentNodeId to enable early filtering of self-messages / 传递 currentNodeId 以启用早期过滤自己的消息
182+
await _messageQueueService.ConsumeAsync(_queueName, HandleMessageAsync, autoAck: false, currentNodeId: _nodeId);
158183

159184
// Wait a bit for initial node discovery / 等待初始节点发现
160185
_logger.LogWarning($"[HybridClusterTransport] 等待初始节点发现... - NodeId: {_nodeId}");
@@ -253,7 +278,7 @@ public async Task SendAsync(string nodeId, ClusterMessage message)
253278

254279
await _messageQueueService.PublishAsync(ExchangeName, routingKey, messageBytes, properties);
255280

256-
_logger.LogWarning($"[HybridClusterTransport] 消息发送成功 - TargetNodeId: {nodeId}, MessageId: {message.MessageId}, MessageType: {message.Type}, CurrentNodeId: {_nodeId}, MessageSize: {messageBytes.Length} bytes");
281+
_logger.LogTrace($"[HybridClusterTransport] 消息发送成功 - TargetNodeId: {nodeId}, MessageId: {message.MessageId}, MessageType: {message.Type}, CurrentNodeId: {_nodeId}, MessageSize: {messageBytes.Length} bytes");
257282
}
258283
catch (Exception ex)
259284
{
@@ -298,12 +323,26 @@ public async Task BroadcastAsync(ClusterMessage message)
298323
{
299324
MessageId = message.MessageId,
300325
CorrelationId = message.MessageId,
301-
Timestamp = message.Timestamp
326+
Timestamp = message.Timestamp,
327+
Headers = new Dictionary<string, object>
328+
{
329+
{ "FromNodeId", _nodeId } // Add FromNodeId header to filter self-messages early / 添加 FromNodeId header 以便早期过滤自己的消息
330+
}
302331
};
303332

304-
await _messageQueueService.PublishAsync(ExchangeName, BroadcastRoutingKey, messageBytes, properties);
305-
306-
_logger.LogWarning($"[HybridClusterTransport] 广播消息成功 - MessageId: {message.MessageId}, MessageType: {message.Type}, CurrentNodeId: {_nodeId}, MessageSize: {messageBytes.Length} bytes, 已知节点数: {_knownNodes.Count}");
333+
// Only broadcast if there are other nodes (excluding self) / 只有在有其他节点时才广播（排除自己）
334+
// Note: _knownNodes does not include self, so _knownNodes.Count is the count of other nodes
335+
// 注意：_knownNodes 不包含自己，所以 _knownNodes.Count 就是其他节点的数量
336+
var otherNodesCount = _knownNodes.Count;
337+
if (otherNodesCount > 0)
338+
{
339+
await _messageQueueService.PublishAsync(ExchangeName, BroadcastRoutingKey, messageBytes, properties);
340+
_logger.LogTrace($"[HybridClusterTransport] 广播消息成功 - MessageId: {message.MessageId}, MessageType: {message.Type}, CurrentNodeId: {_nodeId}, MessageSize: {messageBytes.Length} bytes, 已知节点数: {_knownNodes.Count}, 其他节点数: {otherNodesCount}");
341+
}
342+
else
343+
{
344+
_logger.LogTrace($"[HybridClusterTransport] 跳过广播消息（没有其他节点）- MessageId: {message.MessageId}, MessageType: {message.Type}, CurrentNodeId: {_nodeId}, 已知节点数: {_knownNodes.Count}, 其他节点数: {otherNodesCount}");
345+
}
307346
}
308347
catch (Exception ex)
309348
{
@@ -620,9 +659,9 @@ private async Task<bool> HandleMessageAsync(byte[] body, MessageProperties prope
620659
return true; // Ack to remove from queue / 确认以从队列中移除
621660
}
622661

623-
_logger.LogWarning($"[HybridClusterTransport] 收到集群消息 - MessageType: {message.Type}, FromNodeId: {message.FromNodeId}, ToNodeId: {message.ToNodeId}, MessageId: {message.MessageId}, CurrentNodeId: {_nodeId}, MessageSize: {body.Length} bytes");
662+
_logger.LogTrace($"[HybridClusterTransport] 收到集群消息 - MessageType: {message.Type}, FromNodeId: {message.FromNodeId}, ToNodeId: {message.ToNodeId}, MessageId: {message.MessageId}, CurrentNodeId: {_nodeId}, MessageSize: {body.Length} bytes");
624663

625-
// Ignore messages from self / 忽略来自自己的消息
664+
// Ignore messages from self FIRST to avoid processing own messages / 首先忽略来自自己的消息，避免处理自己的消息
626665
if (message.FromNodeId == _nodeId)
627666
{
628667
_logger.LogWarning($"[HybridClusterTransport] 忽略来自自己的消息 - MessageType: {message.Type}, MessageId: {message.MessageId}, FromNodeId: {message.FromNodeId}, CurrentNodeId: {_nodeId}");
@@ -690,29 +729,25 @@ private async Task<bool> HandleMessageAsync(byte[] body, MessageProperties prope
690729
{
691730
var logData = new
692731
{
693-
location = "HybridClusterTransport.cs:516",
694-
message = "Before invoking MessageReceived event",
695-
data = new
696-
{
697-
messageType = message.Type.ToString(),
698-
fromNodeId = message.FromNodeId,
699-
toNodeId = message.ToNodeId ?? "null",
700-
messageId = message.MessageId,
701-
currentNodeId = _nodeId,
702-
hasSubscribers = MessageReceived != null,
703-
subscriberCount = MessageReceived?.GetInvocationList()?.Length ?? 0
704-
},
705-
timestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(),
706-
sessionId = "debug-session",
707-
runId = "run1",
708-
hypothesisId = "A"
709-
};
710-
var logJson = JsonSerializer.Serialize(logData);
711-
System.IO.File.AppendAllText(@"e:\OneDrive\Work\WorkSpaces\.cursor\debug.log", logJson + Environment.NewLine);
732+
// Targeted message: same message can be sent to different nodes
733+
// 定向消息：相同消息可以发送到不同节点
734+
deduplicationKey = $"{message.MessageId}:{message.FromNodeId}:{message.ToNodeId}";
735+
}
736+
737+
// Check if we've already processed this message / 检查是否已处理过此消息
738+
if (_processedMessageIds.TryGetValue(deduplicationKey, out var processedTime))
739+
{
740+
// Message already processed, ignore it / 消息已处理，忽略它
741+
_logger.LogDebug($"Ignoring duplicate message {message.MessageId} (key: {deduplicationKey}, processed at {processedTime:yyyy-MM-dd HH:mm:ss})");
742+
return true; // Ack to remove from queue / 确认以从队列中移除
743+
}
744+
745+
// Mark message as processed / 标记消息为已处理
746+
_processedMessageIds.TryAdd(deduplicationKey, DateTime.UtcNow);
712747
}
713-
catch { }
714-
// #endregion
715-
748+
749+
// Trigger message received event / 触发消息接收事件
750+
_logger.LogTrace($"[HybridClusterTransport] 触发消息接收事件 - MessageType: {message.Type}, FromNodeId: {message.FromNodeId}, ToNodeId: {message.ToNodeId}, MessageId: {message.MessageId}, CurrentNodeId: {_nodeId}");
716751
MessageReceived?.Invoke(this, new ClusterMessageEventArgs
717752
{
718753
FromNodeId = message.FromNodeId,

‎Cyaim.WebSocketServer/Cluster/Cyaim.WebSocketServer.Cluster.Hybrid/RedisNodeDiscoveryService.cs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -64,8 +64,8 @@ public RedisNodeDiscoveryService(
6464
// Heartbeat every 5 seconds / 每 5 秒发送一次心跳
6565
_heartbeatTimer = new Timer(SendHeartbeat, null, TimeSpan.Zero, TimeSpan.FromSeconds(5));
6666

67-
// Discover nodes every 10 seconds / 每 10 秒发现一次节点
68-
_discoveryTimer = new Timer(DiscoverNodes, null, TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(10));
67+
// Discover nodes every 5 seconds (more frequent to catch nodes faster) / 每 5 秒发现一次节点（更频繁以更快捕获节点）
68+
_discoveryTimer = new Timer(DiscoverNodes, null, TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(5));
6969
}
7070

7171
/// <summary>
@@ -262,7 +262,7 @@ private async void DiscoverNodes(object state)
262262
/// <summary>
263263
/// Discover nodes asynchronously / 异步发现节点
264264
/// </summary>
265-
private async Task DiscoverNodesAsync()
265+
public async Task DiscoverNodesAsync()
266266
{
267267
try
268268
{

‎Cyaim.WebSocketServer/Cyaim.WebSocketServer/Infrastructure/Cluster/RaftNode.cs‎

Lines changed: 34 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -408,23 +408,33 @@ private void SendHeartbeat(object state)
408408

409409
_ = Task.Run(async () =>
410410
{
411-
var appendEntries = new AppendEntriesMessage
411+
try
412412
{
413-
Term = CurrentTerm,
414-
LeaderId = _nodeId,
415-
PrevLogIndex = 0,
416-
PrevLogTerm = 0,
417-
Entries = new RaftLogEntry[0],
418-
LeaderCommit = CommitIndex
419-
};
420-
421-
var message = new ClusterMessage
413+
var appendEntries = new AppendEntriesMessage
414+
{
415+
Term = CurrentTerm,
416+
LeaderId = _nodeId,
417+
PrevLogIndex = 0,
418+
PrevLogTerm = 0,
419+
Entries = new RaftLogEntry[0],
420+
LeaderCommit = CommitIndex
421+
};
422+
423+
var message = new ClusterMessage
424+
{
425+
Type = ClusterMessageType.AppendEntries,
426+
Payload = System.Text.Json.JsonSerializer.Serialize(appendEntries),
427+
// Use a consistent MessageId for heartbeats to enable deduplication / 使用一致的消息ID以便去重
428+
MessageId = $"heartbeat:{_nodeId}:{CurrentTerm}"
429+
};
430+
431+
_logger.LogDebug($"[RaftNode] Sending heartbeat - NodeId: {_nodeId}, Term: {CurrentTerm}, MessageId: {message.MessageId}");
432+
await _transport.BroadcastAsync(message);
433+
}
434+
catch (Exception ex)
422435
{
423-
Type = ClusterMessageType.AppendEntries,
424-
Payload = System.Text.Json.JsonSerializer.Serialize(appendEntries)
425-
};
426-
427-
await _transport.BroadcastAsync(message);
436+
_logger.LogError(ex, $"[RaftNode] Error sending heartbeat - NodeId: {_nodeId}");
437+
}
428438
});
429439
}
430440

@@ -517,6 +527,10 @@ private void HandleRequestVote(ClusterMessage message)
517527
{
518528
_logger.LogWarning($"[RaftNode] 投票拒绝 - NodeId: {_nodeId}, RequestTerm: {request.Term}, CurrentTerm: {CurrentTerm}, VotedFor: {VotedFor}, IsLogUpToDate: {IsLogUpToDate(request.LastLogIndex, request.LastLogTerm)}");
519529
}
530+
else
531+
{
532+
_logger.LogWarning($"[RaftNode] 投票拒绝 - NodeId: {_nodeId}, RequestTerm: {request.Term}, CurrentTerm: {CurrentTerm}, VotedFor: {VotedFor}, IsLogUpToDate: {IsLogUpToDate(request.LastLogIndex, request.LastLogTerm)}");
533+
}
520534

521535
var response = new RequestVoteResponseMessage
522536
{
@@ -541,6 +555,10 @@ private void HandleRequestVote(ClusterMessage message)
541555
{
542556
_logger.LogWarning($"[RaftNode] 投票响应发送成功 - NodeId: {_nodeId}, ToNodeId: {message.FromNodeId}, VoteGranted: {voteGranted}");
543557
}
558+
else
559+
{
560+
_logger.LogWarning($"[RaftNode] 投票响应发送成功 - NodeId: {_nodeId}, ToNodeId: {message.FromNodeId}, VoteGranted: {voteGranted}");
561+
}
544562
});
545563
}
546564
catch (Exception ex)
@@ -899,7 +917,7 @@ private List<string> GetKnownNodeIds()
899917
}
900918
}
901919

902-
_logger.LogDebug($"GetKnownNodeIds for node {_nodeId} returned {nodeIds.Count} node(s): {string.Join(", ", nodeIds)}");
920+
_logger.LogTrace($"GetKnownNodeIds for node {_nodeId} returned {nodeIds.Count} node(s): {string.Join(", ", nodeIds)}");
903921
return nodeIds;
904922
}
905923

0 commit comments

Comments
 (0)