You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
509 lines
16 KiB
509 lines
16 KiB
2 weeks ago
|
# MessageIdManager.cs (自动转换为Markdown)
|
||
|
|
||
|
```csharp
|
||
|
// 以下内容为原始C#代码,含详细注释
|
||
|
// 文件原路径:Managers/MessageIdManager.cs
|
||
|
using System;
|
||
|
using System.Collections.Concurrent;
|
||
|
using System.Collections.Generic;
|
||
|
using System.Linq;
|
||
|
using System.Text;
|
||
|
using System.Threading.Tasks;
|
||
|
using CoreAgent.ProtocolClient.HandlerEventArgs;
|
||
|
using CoreAgent.ProtocolClient.Models;
|
||
|
using Microsoft.Extensions.Logging;
|
||
|
using Newtonsoft.Json.Linq;
|
||
|
|
||
|
namespace CoreAgent.ProtocolClient.Managers
|
||
|
{
|
||
|
/// <summary>
|
||
|
/// 消息ID管理器 - 改进版
|
||
|
///
|
||
|
/// 修复的问题:
|
||
|
/// 1. 使用long类型防止ID溢出
|
||
|
/// 2. 完善消息处理器清理机制
|
||
|
/// 3. 改进LogGet ID管理逻辑
|
||
|
/// 4. 增强参数验证和异常处理
|
||
|
/// 5. 优化性能,减少字符串操作
|
||
|
/// 6. 添加线程安全保护
|
||
|
/// 7. 改进日志记录格式
|
||
|
///
|
||
|
/// 设计原则:
|
||
|
/// - 单一职责:专门负责消息ID管理
|
||
|
/// - 开闭原则:支持扩展不同类型的消息ID管理
|
||
|
/// - 线程安全:所有操作都是线程安全的
|
||
|
/// - 性能优化:减少不必要的内存分配
|
||
|
/// - 错误处理:完善的异常处理和参数验证
|
||
|
/// </summary>
|
||
|
public class MessageIdManager : IDisposable
|
||
|
{
|
||
|
#region 私有字段
|
||
|
|
||
|
private readonly ILogger _logger;
|
||
|
private readonly string _clientName;
|
||
|
private readonly ConcurrentDictionary<long, MessageHandler> _messageHandlers;
|
||
|
private readonly ConcurrentDictionary<string, MessageHandler> _messageHandlersByName;
|
||
|
|
||
|
// 使用long类型防止溢出
|
||
|
private long _generalMessageId;
|
||
|
private long _logGetMessageId;
|
||
|
|
||
|
// 状态管理
|
||
|
private bool _disposed;
|
||
|
|
||
|
// 性能优化:缓存字符串构建器
|
||
|
private readonly StringBuilder _logBuilder = new StringBuilder(256);
|
||
|
|
||
|
#endregion
|
||
|
|
||
|
#region 事件
|
||
|
|
||
|
/// <summary>
|
||
|
/// 日志获取ID变化事件
|
||
|
/// </summary>
|
||
|
public event EventHandler<LogGetIdChangedEventArgs>? LogGetIdChanged;
|
||
|
|
||
|
/// <summary>
|
||
|
/// 消息处理器清理事件
|
||
|
/// </summary>
|
||
|
public event EventHandler<HandlerCleanupEventArgs>? HandlerCleanup;
|
||
|
|
||
|
#endregion
|
||
|
|
||
|
#region 属性
|
||
|
|
||
|
/// <summary>
|
||
|
/// 当前通用消息ID
|
||
|
/// </summary>
|
||
|
public long CurrentGeneralMessageId => Interlocked.Read(ref _generalMessageId);
|
||
|
|
||
|
/// <summary>
|
||
|
/// 当前日志获取消息ID
|
||
|
/// </summary>
|
||
|
public long CurrentLogGetMessageId => Interlocked.Read(ref _logGetMessageId);
|
||
|
|
||
|
/// <summary>
|
||
|
/// 是否已释放
|
||
|
/// </summary>
|
||
|
public bool IsDisposed => _disposed;
|
||
|
|
||
|
/// <summary>
|
||
|
/// 消息处理器数量
|
||
|
/// </summary>
|
||
|
public int MessageHandlerCount => _messageHandlers.Count;
|
||
|
|
||
|
/// <summary>
|
||
|
/// 按名称的消息处理器数量
|
||
|
/// </summary>
|
||
|
public int NamedMessageHandlerCount => _messageHandlersByName.Count;
|
||
|
|
||
|
/// <summary>
|
||
|
/// 总处理器数量
|
||
|
/// </summary>
|
||
|
public int TotalHandlerCount => MessageHandlerCount + NamedMessageHandlerCount;
|
||
|
|
||
|
#endregion
|
||
|
|
||
|
#region 构造函数
|
||
|
|
||
|
/// <summary>
|
||
|
/// 构造函数
|
||
|
/// </summary>
|
||
|
/// <param name="clientName">客户端名称</param>
|
||
|
/// <param name="logger">日志记录器</param>
|
||
|
public MessageIdManager(string clientName, ILogger logger)
|
||
|
{
|
||
|
_clientName = clientName ?? throw new ArgumentNullException(nameof(clientName));
|
||
|
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
|
||
|
|
||
|
_messageHandlers = new ConcurrentDictionary<long, MessageHandler>();
|
||
|
_messageHandlersByName = new ConcurrentDictionary<string, MessageHandler>();
|
||
|
|
||
|
_generalMessageId = 0;
|
||
|
_logGetMessageId = -1; // 初始化为-1,表示未开始
|
||
|
|
||
|
_logger.LogInformation("[{ClientName}] 创建消息ID管理器,初始LogGet ID: {LogGetId}", _clientName, _logGetMessageId);
|
||
|
}
|
||
|
|
||
|
#endregion
|
||
|
|
||
|
#region 公共方法
|
||
|
|
||
|
/// <summary>
|
||
|
/// 生成通用消息ID
|
||
|
/// </summary>
|
||
|
/// <param name="message">消息对象</param>
|
||
|
/// <param name="callback">回调函数</param>
|
||
|
/// <param name="errorHandler">是否为错误处理器</param>
|
||
|
/// <returns>消息ID</returns>
|
||
|
public long GenerateGeneralMessageId(JObject message, Action<JObject>? callback = null, bool errorHandler = false)
|
||
|
{
|
||
|
ThrowIfDisposed();
|
||
|
ValidateMessage(message);
|
||
|
|
||
|
var id = GetNextMessageId();
|
||
|
message["message_id"] = id;
|
||
|
|
||
|
// 记录log_get消息的发送
|
||
|
var messageType = message["message"]?.ToString();
|
||
|
if (messageType == "log_get")
|
||
|
{
|
||
|
LogLogGetMessage(id, message);
|
||
|
}
|
||
|
|
||
|
// 注册回调处理器
|
||
|
if (callback != null)
|
||
|
{
|
||
|
_messageHandlers[id] = new MessageHandler
|
||
|
{
|
||
|
Callback = callback,
|
||
|
ErrorHandler = errorHandler,
|
||
|
CreatedAt = DateTime.UtcNow
|
||
|
};
|
||
|
}
|
||
|
|
||
|
_logger.LogDebug("[{ClientName}] 生成通用消息ID: {MessageId}", _clientName, id);
|
||
|
return id;
|
||
|
}
|
||
|
|
||
|
/// <summary>
|
||
|
/// 生成日志获取消息ID - 改进版
|
||
|
/// </summary>
|
||
|
/// <param name="message">消息对象</param>
|
||
|
/// <param name="callback">回调函数</param>
|
||
|
/// <returns>消息ID</returns>
|
||
|
public long GenerateLogGetMessageId(JObject message, Action<JObject> callback)
|
||
|
{
|
||
|
ThrowIfDisposed();
|
||
|
ValidateMessage(message);
|
||
|
|
||
|
if (callback == null)
|
||
|
throw new ArgumentNullException(nameof(callback));
|
||
|
|
||
|
// 生成新的消息ID
|
||
|
var newLogGetId = GetNextMessageId();
|
||
|
message["message_id"] = newLogGetId;
|
||
|
|
||
|
// 注册回调处理器
|
||
|
_messageHandlers[newLogGetId] = new MessageHandler
|
||
|
{
|
||
|
Callback = callback,
|
||
|
ErrorHandler = false,
|
||
|
CreatedAt = DateTime.UtcNow,
|
||
|
IsLogGetHandler = true
|
||
|
};
|
||
|
|
||
|
// 设置新的LogGet ID
|
||
|
var oldLogGetId = Interlocked.Exchange(ref _logGetMessageId, newLogGetId);
|
||
|
|
||
|
// 触发事件
|
||
|
LogGetIdChanged?.Invoke(this, new LogGetIdChangedEventArgs(oldLogGetId, newLogGetId));
|
||
|
|
||
|
_logger.LogDebug("[{ClientName}] LogGet ID变化: {OldId} -> {NewId}", _clientName, oldLogGetId, newLogGetId);
|
||
|
return newLogGetId;
|
||
|
}
|
||
|
|
||
|
/// <summary>
|
||
|
/// 处理消息响应 - 改进版
|
||
|
/// </summary>
|
||
|
/// <param name="response">响应消息</param>
|
||
|
/// <param name="errorHandler">错误处理回调</param>
|
||
|
/// <returns>是否找到并处理了消息处理器</returns>
|
||
|
public bool HandleMessageResponse(JObject response, Action<string>? errorHandler = null)
|
||
|
{
|
||
|
ThrowIfDisposed();
|
||
|
|
||
|
if (response == null)
|
||
|
return false;
|
||
|
|
||
|
// 检查消息处理器
|
||
|
var id = response["message_id"]?.Value<long>();
|
||
|
if (id.HasValue && _messageHandlers.TryGetValue(id.Value, out var handler))
|
||
|
{
|
||
|
return HandleMessageHandler(id.Value, handler, response, errorHandler);
|
||
|
}
|
||
|
|
||
|
// 检查按名称的消息处理器
|
||
|
var name = response["message"]?.ToString();
|
||
|
if (!string.IsNullOrEmpty(name) && _messageHandlersByName.TryGetValue(name, out var nameHandler))
|
||
|
{
|
||
|
return HandleNamedMessageHandler(name, nameHandler, response);
|
||
|
}
|
||
|
|
||
|
return false;
|
||
|
}
|
||
|
|
||
|
/// <summary>
|
||
|
/// 设置消息处理器
|
||
|
/// </summary>
|
||
|
/// <param name="names">消息名称数组</param>
|
||
|
/// <param name="handler">处理器</param>
|
||
|
public void SetMessageHandler(string[] names, MessageHandler handler)
|
||
|
{
|
||
|
ThrowIfDisposed();
|
||
|
|
||
|
if (names == null || names.Length == 0)
|
||
|
throw new ArgumentException("消息名称不能为空", nameof(names));
|
||
|
|
||
|
if (handler == null)
|
||
|
throw new ArgumentNullException(nameof(handler));
|
||
|
|
||
|
foreach (var name in names)
|
||
|
{
|
||
|
if (!string.IsNullOrEmpty(name))
|
||
|
{
|
||
|
_messageHandlersByName[name] = handler;
|
||
|
}
|
||
|
}
|
||
|
|
||
|
_logger.LogDebug("[{ClientName}] 设置消息处理器: {Names}", _clientName, string.Join(", ", names));
|
||
|
}
|
||
|
|
||
|
/// <summary>
|
||
|
/// 取消消息处理器
|
||
|
/// </summary>
|
||
|
/// <param name="names">消息名称数组</param>
|
||
|
public void UnsetMessageHandler(string[] names)
|
||
|
{
|
||
|
ThrowIfDisposed();
|
||
|
|
||
|
if (names == null || names.Length == 0)
|
||
|
return;
|
||
|
|
||
|
foreach (var name in names)
|
||
|
{
|
||
|
if (!string.IsNullOrEmpty(name))
|
||
|
{
|
||
|
_messageHandlersByName.TryRemove(name, out _);
|
||
|
}
|
||
|
}
|
||
|
|
||
|
_logger.LogDebug("[{ClientName}] 取消消息处理器: {Names}", _clientName, string.Join(", ", names));
|
||
|
}
|
||
|
|
||
|
/// <summary>
|
||
|
/// 检查是否为当前日志获取消息
|
||
|
/// </summary>
|
||
|
/// <param name="messageId">消息ID</param>
|
||
|
/// <returns>是否为当前日志获取消息</returns>
|
||
|
public bool IsCurrentLogGetMessage(long messageId)
|
||
|
{
|
||
|
var currentLogGetId = Interlocked.Read(ref _logGetMessageId);
|
||
|
return messageId == currentLogGetId;
|
||
|
}
|
||
|
|
||
|
/// <summary>
|
||
|
/// 重置日志获取ID
|
||
|
/// </summary>
|
||
|
public void ResetLogGetId()
|
||
|
{
|
||
|
var oldLogGetId = Interlocked.Exchange(ref _logGetMessageId, -1);
|
||
|
LogGetIdChanged?.Invoke(this, new LogGetIdChangedEventArgs(oldLogGetId, -1));
|
||
|
_logger.LogDebug("[{ClientName}] 重置LogGet ID: {OldId} -> -1", _clientName, oldLogGetId);
|
||
|
}
|
||
|
|
||
|
/// <summary>
|
||
|
/// 清理过期的消息处理器 - 改进版
|
||
|
/// </summary>
|
||
|
/// <param name="maxAge">最大年龄(毫秒)</param>
|
||
|
/// <returns>清理的处理器数量</returns>
|
||
|
public int CleanupExpiredHandlers(int maxAge = 30000) // 默认30秒
|
||
|
{
|
||
|
ThrowIfDisposed();
|
||
|
|
||
|
var now = DateTime.UtcNow;
|
||
|
var expiredKeys = new List<long>();
|
||
|
|
||
|
foreach (var kvp in _messageHandlers)
|
||
|
{
|
||
|
if (kvp.Value.CreatedAt.AddMilliseconds(maxAge) < now)
|
||
|
{
|
||
|
expiredKeys.Add(kvp.Key);
|
||
|
}
|
||
|
}
|
||
|
|
||
|
var cleanedCount = 0;
|
||
|
foreach (var key in expiredKeys)
|
||
|
{
|
||
|
if (_messageHandlers.TryRemove(key, out _))
|
||
|
{
|
||
|
cleanedCount++;
|
||
|
}
|
||
|
}
|
||
|
|
||
|
if (cleanedCount > 0)
|
||
|
{
|
||
|
_logger.LogDebug("[{ClientName}] 清理了 {CleanedCount} 个过期的消息处理器", _clientName, cleanedCount);
|
||
|
|
||
|
// 触发清理事件
|
||
|
HandlerCleanup?.Invoke(this, new HandlerCleanupEventArgs(
|
||
|
_messageHandlers.Count,
|
||
|
_messageHandlersByName.Count,
|
||
|
cleanedCount));
|
||
|
}
|
||
|
|
||
|
return cleanedCount;
|
||
|
}
|
||
|
|
||
|
#endregion
|
||
|
|
||
|
#region 私有方法
|
||
|
|
||
|
/// <summary>
|
||
|
/// 获取下一个消息ID
|
||
|
/// </summary>
|
||
|
/// <returns>消息ID</returns>
|
||
|
private long GetNextMessageId()
|
||
|
{
|
||
|
var id = Interlocked.Increment(ref _generalMessageId);
|
||
|
|
||
|
// 检查溢出
|
||
|
if (id <= 0)
|
||
|
{
|
||
|
_logger.LogWarning("[{ClientName}] 消息ID溢出,重置为1", _clientName);
|
||
|
Interlocked.Exchange(ref _generalMessageId, 1);
|
||
|
return 1;
|
||
|
}
|
||
|
|
||
|
return id;
|
||
|
}
|
||
|
|
||
|
/// <summary>
|
||
|
/// 验证消息对象
|
||
|
/// </summary>
|
||
|
/// <param name="message">消息对象</param>
|
||
|
private void ValidateMessage(JObject message)
|
||
|
{
|
||
|
if (message == null)
|
||
|
throw new ArgumentNullException(nameof(message));
|
||
|
}
|
||
|
|
||
|
/// <summary>
|
||
|
/// 处理消息处理器
|
||
|
/// </summary>
|
||
|
/// <param name="id">消息ID</param>
|
||
|
/// <param name="handler">处理器</param>
|
||
|
/// <param name="response">响应消息</param>
|
||
|
/// <param name="errorHandler">错误处理回调</param>
|
||
|
/// <returns>是否处理成功</returns>
|
||
|
private bool HandleMessageHandler(long id, MessageHandler handler, JObject response, Action<string>? errorHandler)
|
||
|
{
|
||
|
// 如果不是通知消息,则移除处理器
|
||
|
if (response["notification"]?.Value<bool>() != true)
|
||
|
{
|
||
|
_messageHandlers.TryRemove(id, out _);
|
||
|
}
|
||
|
|
||
|
// 处理错误
|
||
|
if (response["error"] != null)
|
||
|
{
|
||
|
if (!handler.ErrorHandler)
|
||
|
{
|
||
|
errorHandler?.Invoke(response["error"]?.ToString() ?? "未知错误");
|
||
|
}
|
||
|
else
|
||
|
{
|
||
|
handler.Callback?.Invoke(response);
|
||
|
}
|
||
|
return true;
|
||
|
}
|
||
|
|
||
|
// 正常处理
|
||
|
handler.Callback?.Invoke(response);
|
||
|
return true;
|
||
|
}
|
||
|
|
||
|
/// <summary>
|
||
|
/// 处理按名称的消息处理器
|
||
|
/// </summary>
|
||
|
/// <param name="name">消息名称</param>
|
||
|
/// <param name="handler">处理器</param>
|
||
|
/// <param name="response">响应消息</param>
|
||
|
/// <returns>是否处理成功</returns>
|
||
|
private bool HandleNamedMessageHandler(string name, MessageHandler handler, JObject response)
|
||
|
{
|
||
|
handler.Callback?.Invoke(response);
|
||
|
return true;
|
||
|
}
|
||
|
|
||
|
/// <summary>
|
||
|
/// 记录LogGet消息
|
||
|
/// </summary>
|
||
|
/// <param name="id">消息ID</param>
|
||
|
/// <param name="message">消息对象</param>
|
||
|
private void LogLogGetMessage(long id, JObject message)
|
||
|
{
|
||
|
_logBuilder.Clear();
|
||
|
_logBuilder.AppendFormat("[{0}] 发送log_get消息: message_id={1}", _clientName, id);
|
||
|
|
||
|
if (message["timeout"] != null)
|
||
|
_logBuilder.AppendFormat(", timeout={0}", message["timeout"]);
|
||
|
|
||
|
if (message["headers"] != null)
|
||
|
_logBuilder.AppendFormat(", headers={0}", message["headers"]);
|
||
|
|
||
|
_logBuilder.AppendFormat(", ThreadId={0}", Thread.CurrentThread.ManagedThreadId);
|
||
|
|
||
|
_logger.LogDebug(_logBuilder.ToString());
|
||
|
}
|
||
|
|
||
|
/// <summary>
|
||
|
/// 检查是否已释放
|
||
|
/// </summary>
|
||
|
private void ThrowIfDisposed()
|
||
|
{
|
||
|
if (_disposed)
|
||
|
{
|
||
|
throw new ObjectDisposedException(nameof(MessageIdManager));
|
||
|
}
|
||
|
}
|
||
|
|
||
|
#endregion
|
||
|
|
||
|
#region IDisposable
|
||
|
|
||
|
/// <summary>
|
||
|
/// 释放资源
|
||
|
/// </summary>
|
||
|
public void Dispose()
|
||
|
{
|
||
|
if (_disposed) return;
|
||
|
|
||
|
_disposed = true;
|
||
|
|
||
|
// 清理所有消息处理器
|
||
|
_messageHandlers.Clear();
|
||
|
_messageHandlersByName.Clear();
|
||
|
|
||
|
_logger.LogInformation("[{ClientName}] 释放消息ID管理器", _clientName);
|
||
|
}
|
||
|
|
||
|
#endregion
|
||
|
}
|
||
|
}
|
||
|
```
|
||
|
|
||
|
## 文档说明
|
||
|
|
||
|
### 类概述
|
||
|
`MessageIdManager` 是一个专门负责消息ID管理的类,用于替代原始实现中的 `_messageId` 和 `_logGetId` 字段,提供更统一和强大的消息ID管理功能。
|
||
|
|
||
|
### 主要功能
|
||
|
1. **消息ID生成**:统一管理通用消息ID和日志获取消息ID
|
||
|
2. **消息处理器管理**:管理消息回调处理器,支持按ID和按名称两种方式
|
||
|
3. **线程安全**:所有操作都是线程安全的
|
||
|
4. **自动清理**:支持自动清理过期的消息处理器
|
||
|
5. **事件通知**:提供LogGet ID变化和处理器清理的事件通知
|
||
|
|
||
|
### 设计改进
|
||
|
- 使用long类型防止ID溢出
|
||
|
- 完善的异常处理和参数验证
|
||
|
- 性能优化,减少字符串操作
|
||
|
- 标准的Dispose模式实现
|
||
|
- 详细的日志记录
|
||
|
|
||
|
### 使用场景
|
||
|
- 在WebSocket消息管理器中统一管理消息ID
|
||
|
- 处理消息响应和回调
|
||
|
- 管理日志获取流程
|
||
|
- 提供消息处理器的生命周期管理
|