hyh
2 months ago
13 changed files with 1079 additions and 73 deletions
@ -0,0 +1,159 @@ |
|||
using System; |
|||
using System.Diagnostics; |
|||
using System.Threading; |
|||
using System.Threading.Tasks; |
|||
using CellularManagement.WebSocket.Models; |
|||
using Microsoft.Extensions.Logging; |
|||
|
|||
namespace CellularManagement.WebSocket.Handlers |
|||
{ |
|||
/// <summary>
|
|||
/// 基础消息处理器
|
|||
/// </summary>
|
|||
public abstract class BaseMessageHandler : IMessageHandler |
|||
{ |
|||
protected readonly ILogger _logger; |
|||
protected readonly HandlerMetrics _metrics; |
|||
protected readonly Stopwatch _stopwatch; |
|||
private readonly Timer _healthCheckTimer; |
|||
|
|||
protected BaseMessageHandler(ILogger logger) |
|||
{ |
|||
_logger = logger; |
|||
_metrics = new HandlerMetrics |
|||
{ |
|||
HandlerName = GetType().Name, |
|||
HealthStatus = HandlerHealthStatus.Healthy, |
|||
Configuration = new HandlerConfiguration() |
|||
}; |
|||
_stopwatch = new Stopwatch(); |
|||
_healthCheckTimer = new Timer(PerformHealthCheck, null, |
|||
TimeSpan.Zero, |
|||
TimeSpan.FromSeconds(_metrics.Configuration.HealthCheckInterval)); |
|||
} |
|||
|
|||
public abstract int Priority { get; } |
|||
public abstract string Name { get; } |
|||
|
|||
public async Task<WebSocketMessage> HandleAsync(WebSocketMessage message, CancellationToken cancellationToken) |
|||
{ |
|||
if (!_metrics.Configuration.IsEnabled) |
|||
{ |
|||
_logger.LogWarning("处理器已禁用，处理器：{HandlerName}，连接ID：{ConnectionId}", |
|||
Name, message.ConnectionId); |
|||
throw new InvalidOperationException($"处理器 {Name} 已禁用"); |
|||
} |
|||
|
|||
_metrics.TotalMessages++; |
|||
_stopwatch.Restart(); |
|||
|
|||
try |
|||
{ |
|||
var result = await ProcessMessageAsync(message, cancellationToken); |
|||
_metrics.SuccessMessages++; |
|||
return result; |
|||
} |
|||
catch (Exception ex) |
|||
{ |
|||
_metrics.FailedMessages++; |
|||
_metrics.HealthCheckFailureCount++; |
|||
_logger.LogError(ex, "消息处理失败，处理器：{HandlerName}，连接ID：{ConnectionId}", |
|||
Name, message.ConnectionId); |
|||
throw; |
|||
} |
|||
finally |
|||
{ |
|||
_stopwatch.Stop(); |
|||
_metrics.LastProcessedTime = DateTime.UtcNow; |
|||
UpdateAverageProcessingTime(); |
|||
} |
|||
} |
|||
|
|||
protected abstract Task<WebSocketMessage> ProcessMessageAsync(WebSocketMessage message, CancellationToken cancellationToken); |
|||
|
|||
public HandlerStatistics GetStatistics() |
|||
{ |
|||
return new HandlerStatistics |
|||
{ |
|||
TotalMessages = _metrics.TotalMessages, |
|||
SuccessMessages = _metrics.SuccessMessages, |
|||
FailedMessages = _metrics.FailedMessages, |
|||
AverageProcessingTime = _metrics.AverageProcessingTime, |
|||
LastProcessedTime = _metrics.LastProcessedTime |
|||
}; |
|||
} |
|||
|
|||
/// <summary>
|
|||
/// 获取处理器指标
|
|||
/// </summary>
|
|||
public HandlerMetrics GetMetrics() |
|||
{ |
|||
return _metrics; |
|||
} |
|||
|
|||
/// <summary>
|
|||
/// 更新处理器配置
|
|||
/// </summary>
|
|||
public void UpdateConfiguration(HandlerConfiguration configuration) |
|||
{ |
|||
_metrics.Configuration = configuration; |
|||
_healthCheckTimer.Change( |
|||
TimeSpan.Zero, |
|||
TimeSpan.FromSeconds(configuration.HealthCheckInterval)); |
|||
_logger.LogInformation("更新处理器配置，处理器：{HandlerName}，配置：{Configuration}", |
|||
Name, System.Text.Json.JsonSerializer.Serialize(configuration)); |
|||
} |
|||
|
|||
private void UpdateAverageProcessingTime() |
|||
{ |
|||
if (_metrics.TotalMessages == 1) |
|||
{ |
|||
_metrics.AverageProcessingTime = _stopwatch.ElapsedMilliseconds; |
|||
} |
|||
else |
|||
{ |
|||
_metrics.AverageProcessingTime = (_metrics.AverageProcessingTime * (_metrics.TotalMessages - 1) + |
|||
_stopwatch.ElapsedMilliseconds) / _metrics.TotalMessages; |
|||
} |
|||
} |
|||
|
|||
private void PerformHealthCheck(object state) |
|||
{ |
|||
try |
|||
{ |
|||
_metrics.LastHealthCheckTime = DateTime.UtcNow; |
|||
|
|||
// 检查失败次数
|
|||
if (_metrics.HealthCheckFailureCount >= _metrics.Configuration.MaxFailureThreshold) |
|||
{ |
|||
_metrics.HealthStatus = HandlerHealthStatus.Unhealthy; |
|||
_logger.LogWarning("处理器健康检查失败，处理器：{HandlerName}，失败次数：{FailureCount}", |
|||
Name, _metrics.HealthCheckFailureCount); |
|||
return; |
|||
} |
|||
|
|||
// 检查平均处理时间
|
|||
if (_metrics.AverageProcessingTime > _metrics.Configuration.AverageProcessingTimeWarningThreshold) |
|||
{ |
|||
_metrics.HealthStatus = HandlerHealthStatus.Warning; |
|||
_logger.LogWarning("处理器性能警告，处理器：{HandlerName}，平均处理时间：{AverageTime}ms", |
|||
Name, _metrics.AverageProcessingTime); |
|||
return; |
|||
} |
|||
|
|||
_metrics.HealthStatus = HandlerHealthStatus.Healthy; |
|||
_metrics.HealthCheckFailureCount = 0; |
|||
_logger.LogDebug("处理器健康检查通过，处理器：{HandlerName}", Name); |
|||
} |
|||
catch (Exception ex) |
|||
{ |
|||
_logger.LogError(ex, "处理器健康检查异常，处理器：{HandlerName}", Name); |
|||
} |
|||
} |
|||
|
|||
public void Dispose() |
|||
{ |
|||
_healthCheckTimer?.Dispose(); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,227 @@ |
|||
using System; |
|||
using System.Collections.Concurrent; |
|||
using System.Collections.Generic; |
|||
using System.Linq; |
|||
using System.Threading; |
|||
using System.Threading.Tasks; |
|||
using CellularManagement.WebSocket.Models; |
|||
using Microsoft.Extensions.Logging; |
|||
|
|||
namespace CellularManagement.WebSocket.Handlers |
|||
{ |
|||
/// <summary>
|
|||
/// 处理器管理器
|
|||
/// </summary>
|
|||
public class HandlerManager : IDisposable |
|||
{ |
|||
private readonly ILogger<HandlerManager> _logger; |
|||
private readonly ConcurrentDictionary<string, List<IMessageHandler>> _handlers; |
|||
private readonly ConcurrentDictionary<string, HandlerMetrics> _handlerMetrics; |
|||
private readonly ConcurrentDictionary<string, SemaphoreSlim> _handlerLocks; |
|||
private readonly ConcurrentDictionary<string, int> _handlerPoolSizes; |
|||
private readonly Timer _performanceSamplingTimer; |
|||
private readonly object _handlersLock = new object(); |
|||
|
|||
/// <summary>
|
|||
/// 处理器状态变更事件
|
|||
/// </summary>
|
|||
public event EventHandler<HandlerStatusChangedEventArgs> HandlerStatusChanged; |
|||
|
|||
/// <summary>
|
|||
/// 处理器性能采样事件
|
|||
/// </summary>
|
|||
public event EventHandler<HandlerPerformanceSampleEventArgs> PerformanceSampled; |
|||
|
|||
public HandlerManager(ILogger<HandlerManager> logger) |
|||
{ |
|||
_logger = logger; |
|||
_handlers = new ConcurrentDictionary<string, List<IMessageHandler>>(); |
|||
_handlerMetrics = new ConcurrentDictionary<string, HandlerMetrics>(); |
|||
_handlerLocks = new ConcurrentDictionary<string, SemaphoreSlim>(); |
|||
_handlerPoolSizes = new ConcurrentDictionary<string, int>(); |
|||
_performanceSamplingTimer = new Timer(CollectPerformanceSamples, null, |
|||
TimeSpan.Zero, TimeSpan.FromSeconds(5)); |
|||
} |
|||
|
|||
/// <summary>
|
|||
/// 注册处理器
|
|||
/// </summary>
|
|||
public async Task RegisterHandlerAsync(string messageType, IMessageHandler handler, int poolSize = 1) |
|||
{ |
|||
var handlerLock = _handlerLocks.GetOrAdd(handler.Name, _ => new SemaphoreSlim(1, 1)); |
|||
await handlerLock.WaitAsync(); |
|||
|
|||
try |
|||
{ |
|||
_logger.LogInformation("注册处理器，消息类型：{MessageType}，处理器：{HandlerName}，池大小：{PoolSize}", |
|||
messageType, handler.Name, poolSize); |
|||
|
|||
if (!_handlers.ContainsKey(messageType)) |
|||
{ |
|||
_handlers[messageType] = new List<IMessageHandler>(); |
|||
} |
|||
|
|||
// 检查处理器是否已存在
|
|||
if (_handlers[messageType].Any(h => h.Name == handler.Name)) |
|||
{ |
|||
throw new InvalidOperationException($"处理器 {handler.Name} 已注册"); |
|||
} |
|||
|
|||
_handlers[messageType].Add(handler); |
|||
_handlers[messageType].Sort((a, b) => a.Priority.CompareTo(b.Priority)); |
|||
_handlerMetrics[handler.Name] = handler.GetMetrics(); |
|||
_handlerPoolSizes[handler.Name] = poolSize; |
|||
|
|||
_logger.LogDebug("处理器注册完成，当前处理器数量：{HandlerCount}", _handlers[messageType].Count); |
|||
} |
|||
finally |
|||
{ |
|||
handlerLock.Release(); |
|||
} |
|||
} |
|||
|
|||
/// <summary>
|
|||
/// 移除处理器
|
|||
/// </summary>
|
|||
public async Task RemoveHandlerAsync(string messageType, string handlerName) |
|||
{ |
|||
var handlerLock = _handlerLocks.GetOrAdd(handlerName, _ => new SemaphoreSlim(1, 1)); |
|||
await handlerLock.WaitAsync(); |
|||
|
|||
try |
|||
{ |
|||
_logger.LogInformation("移除处理器，消息类型：{MessageType}，处理器：{HandlerName}", |
|||
messageType, handlerName); |
|||
|
|||
if (_handlers.TryGetValue(messageType, out var handlers)) |
|||
{ |
|||
var handler = handlers.FirstOrDefault(h => h.Name == handlerName); |
|||
if (handler != null) |
|||
{ |
|||
handlers.Remove(handler); |
|||
_handlerMetrics.TryRemove(handlerName, out _); |
|||
_handlerPoolSizes.TryRemove(handlerName, out _); |
|||
handler.Dispose(); |
|||
} |
|||
} |
|||
} |
|||
finally |
|||
{ |
|||
handlerLock.Release(); |
|||
} |
|||
} |
|||
|
|||
/// <summary>
|
|||
/// 获取处理器
|
|||
/// </summary>
|
|||
public async Task<IMessageHandler> GetHandlerAsync(string messageType, string handlerName) |
|||
{ |
|||
var handlerLock = _handlerLocks.GetOrAdd(handlerName, _ => new SemaphoreSlim(1, 1)); |
|||
await handlerLock.WaitAsync(); |
|||
|
|||
try |
|||
{ |
|||
if (_handlers.TryGetValue(messageType, out var handlers)) |
|||
{ |
|||
return handlers.FirstOrDefault(h => h.Name == handlerName); |
|||
} |
|||
return null; |
|||
} |
|||
finally |
|||
{ |
|||
handlerLock.Release(); |
|||
} |
|||
} |
|||
|
|||
/// <summary>
|
|||
/// 获取所有处理器
|
|||
/// </summary>
|
|||
public async Task<List<IMessageHandler>> GetAllHandlersAsync(string messageType) |
|||
{ |
|||
if (_handlers.TryGetValue(messageType, out var handlers)) |
|||
{ |
|||
return new List<IMessageHandler>(handlers); |
|||
} |
|||
return new List<IMessageHandler>(); |
|||
} |
|||
|
|||
/// <summary>
|
|||
/// 更新处理器配置
|
|||
/// </summary>
|
|||
public async Task UpdateHandlerConfigurationAsync(string handlerName, HandlerConfiguration configuration) |
|||
{ |
|||
var handlerLock = _handlerLocks.GetOrAdd(handlerName, _ => new SemaphoreSlim(1, 1)); |
|||
await handlerLock.WaitAsync(); |
|||
|
|||
try |
|||
{ |
|||
var handler = _handlers.Values |
|||
.SelectMany(h => h) |
|||
.FirstOrDefault(h => h.Name == handlerName); |
|||
|
|||
if (handler == null) |
|||
{ |
|||
throw new InvalidOperationException($"未找到处理器：{handlerName}"); |
|||
} |
|||
|
|||
handler.UpdateConfiguration(configuration); |
|||
_handlerMetrics[handlerName] = handler.GetMetrics(); |
|||
_logger.LogInformation("更新处理器配置，处理器：{HandlerName}，配置：{Configuration}", |
|||
handlerName, System.Text.Json.JsonSerializer.Serialize(configuration)); |
|||
} |
|||
finally |
|||
{ |
|||
handlerLock.Release(); |
|||
} |
|||
} |
|||
|
|||
/// <summary>
|
|||
/// 获取处理器指标
|
|||
/// </summary>
|
|||
public Dictionary<string, HandlerMetrics> GetHandlerMetrics() |
|||
{ |
|||
return _handlerMetrics.ToDictionary(kvp => kvp.Key, kvp => kvp.Value); |
|||
} |
|||
|
|||
/// <summary>
|
|||
/// 更新处理器状态
|
|||
/// </summary>
|
|||
public void UpdateHandlerStatus(string handlerName, HandlerHealthStatus newStatus, string reason) |
|||
{ |
|||
if (_handlerMetrics.TryGetValue(handlerName, out var metrics)) |
|||
{ |
|||
var oldStatus = metrics.HealthStatus; |
|||
metrics.HealthStatus = newStatus; |
|||
|
|||
HandlerStatusChanged?.Invoke(this, new HandlerStatusChangedEventArgs( |
|||
handlerName, |
|||
metrics.MessageType, |
|||
oldStatus, |
|||
newStatus, |
|||
reason)); |
|||
} |
|||
} |
|||
|
|||
private void CollectPerformanceSamples(object state) |
|||
{ |
|||
foreach (var kvp in _handlerMetrics) |
|||
{ |
|||
var metrics = kvp.Value; |
|||
PerformanceSampled?.Invoke(this, new HandlerPerformanceSampleEventArgs( |
|||
metrics.HandlerName, |
|||
metrics.MessageType, |
|||
metrics.AverageProcessingTime, |
|||
metrics.SuccessMessages > 0 && metrics.FailedMessages == 0)); |
|||
} |
|||
} |
|||
|
|||
public void Dispose() |
|||
{ |
|||
_performanceSamplingTimer?.Dispose(); |
|||
foreach (var handlerLock in _handlerLocks.Values) |
|||
{ |
|||
handlerLock.Dispose(); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,40 @@ |
|||
using CellularManagement.WebSocket.Connection; |
|||
using CellularManagement.WebSocket.Models; |
|||
using CellularManagement.WebSocket.Services; |
|||
using Microsoft.Extensions.Logging; |
|||
|
|||
namespace CellularManagement.WebSocket.Handlers |
|||
{ |
|||
/// <summary>
|
|||
/// WebSocket 消息处理器适配器
|
|||
/// </summary>
|
|||
public class WebSocketMessageHandlerAdapter : BaseMessageHandler |
|||
{ |
|||
private readonly IWebSocketMessageHandler _handler; |
|||
|
|||
public WebSocketMessageHandlerAdapter( |
|||
IWebSocketMessageHandler handler, |
|||
ILogger<WebSocketMessageService> logger) |
|||
: base(logger) |
|||
{ |
|||
_handler = handler; |
|||
} |
|||
|
|||
public override string Name => _handler.GetType().Name; |
|||
public override int Priority => 0; |
|||
|
|||
protected override async Task<WebSocketMessage> ProcessMessageAsync(WebSocketMessage message, CancellationToken cancellationToken) |
|||
{ |
|||
try |
|||
{ |
|||
var result = await _handler.HandleAsync(message); |
|||
return result; |
|||
} |
|||
catch (Exception ex) |
|||
{ |
|||
_logger.LogError(ex, "处理消息时发生错误，处理器：{HandlerName}", Name); |
|||
throw; |
|||
} |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,100 @@ |
|||
using System; |
|||
|
|||
namespace CellularManagement.WebSocket.Models |
|||
{ |
|||
/// <summary>
|
|||
/// 处理器事件参数
|
|||
/// </summary>
|
|||
public class HandlerEventArgs : EventArgs |
|||
{ |
|||
/// <summary>
|
|||
/// 处理器名称
|
|||
/// </summary>
|
|||
public string HandlerName { get; } |
|||
|
|||
/// <summary>
|
|||
/// 消息类型
|
|||
/// </summary>
|
|||
public string MessageType { get; } |
|||
|
|||
/// <summary>
|
|||
/// 事件时间
|
|||
/// </summary>
|
|||
public DateTime EventTime { get; } |
|||
|
|||
public HandlerEventArgs(string handlerName, string messageType) |
|||
{ |
|||
HandlerName = handlerName; |
|||
MessageType = messageType; |
|||
EventTime = DateTime.UtcNow; |
|||
} |
|||
} |
|||
|
|||
/// <summary>
|
|||
/// 处理器状态变更事件参数
|
|||
/// </summary>
|
|||
public class HandlerStatusChangedEventArgs : HandlerEventArgs |
|||
{ |
|||
/// <summary>
|
|||
/// 旧状态
|
|||
/// </summary>
|
|||
public HandlerHealthStatus OldStatus { get; } |
|||
|
|||
/// <summary>
|
|||
/// 新状态
|
|||
/// </summary>
|
|||
public HandlerHealthStatus NewStatus { get; } |
|||
|
|||
/// <summary>
|
|||
/// 变更原因
|
|||
/// </summary>
|
|||
public string Reason { get; } |
|||
|
|||
public HandlerStatusChangedEventArgs( |
|||
string handlerName, |
|||
string messageType, |
|||
HandlerHealthStatus oldStatus, |
|||
HandlerHealthStatus newStatus, |
|||
string reason) |
|||
: base(handlerName, messageType) |
|||
{ |
|||
OldStatus = oldStatus; |
|||
NewStatus = newStatus; |
|||
Reason = reason; |
|||
} |
|||
} |
|||
|
|||
/// <summary>
|
|||
/// 处理器性能采样事件参数
|
|||
/// </summary>
|
|||
public class HandlerPerformanceSampleEventArgs : HandlerEventArgs |
|||
{ |
|||
/// <summary>
|
|||
/// 处理时间（毫秒）
|
|||
/// </summary>
|
|||
public double ProcessingTime { get; } |
|||
|
|||
/// <summary>
|
|||
/// 是否成功
|
|||
/// </summary>
|
|||
public bool IsSuccess { get; } |
|||
|
|||
/// <summary>
|
|||
/// 错误信息
|
|||
/// </summary>
|
|||
public string ErrorMessage { get; } |
|||
|
|||
public HandlerPerformanceSampleEventArgs( |
|||
string handlerName, |
|||
string messageType, |
|||
double processingTime, |
|||
bool isSuccess, |
|||
string errorMessage = null) |
|||
: base(handlerName, messageType) |
|||
{ |
|||
ProcessingTime = processingTime; |
|||
IsSuccess = isSuccess; |
|||
ErrorMessage = errorMessage; |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,122 @@ |
|||
using System; |
|||
|
|||
namespace CellularManagement.WebSocket.Models |
|||
{ |
|||
/// <summary>
|
|||
/// 处理器性能指标
|
|||
/// </summary>
|
|||
public class HandlerMetrics |
|||
{ |
|||
/// <summary>
|
|||
/// 处理器名称
|
|||
/// </summary>
|
|||
public string HandlerName { get; set; } |
|||
|
|||
/// <summary>
|
|||
/// 消息类型
|
|||
/// </summary>
|
|||
public string MessageType { get; set; } |
|||
|
|||
/// <summary>
|
|||
/// 处理消息总数
|
|||
/// </summary>
|
|||
public long TotalMessages { get; set; } |
|||
|
|||
/// <summary>
|
|||
/// 成功处理消息数
|
|||
/// </summary>
|
|||
public long SuccessMessages { get; set; } |
|||
|
|||
/// <summary>
|
|||
/// 失败消息数
|
|||
/// </summary>
|
|||
public long FailedMessages { get; set; } |
|||
|
|||
/// <summary>
|
|||
/// 平均处理时间（毫秒）
|
|||
/// </summary>
|
|||
public double AverageProcessingTime { get; set; } |
|||
|
|||
/// <summary>
|
|||
/// 最后处理时间
|
|||
/// </summary>
|
|||
public DateTime LastProcessedTime { get; set; } |
|||
|
|||
/// <summary>
|
|||
/// 处理器状态
|
|||
/// </summary>
|
|||
public HandlerHealthStatus HealthStatus { get; set; } |
|||
|
|||
/// <summary>
|
|||
/// 最后健康检查时间
|
|||
/// </summary>
|
|||
public DateTime LastHealthCheckTime { get; set; } |
|||
|
|||
/// <summary>
|
|||
/// 健康检查失败次数
|
|||
/// </summary>
|
|||
public int HealthCheckFailureCount { get; set; } |
|||
|
|||
/// <summary>
|
|||
/// 处理器配置
|
|||
/// </summary>
|
|||
public HandlerConfiguration Configuration { get; set; } |
|||
} |
|||
|
|||
/// <summary>
|
|||
/// 处理器健康状态
|
|||
/// </summary>
|
|||
public enum HandlerHealthStatus |
|||
{ |
|||
/// <summary>
|
|||
/// 健康
|
|||
/// </summary>
|
|||
Healthy, |
|||
|
|||
/// <summary>
|
|||
/// 警告
|
|||
/// </summary>
|
|||
Warning, |
|||
|
|||
/// <summary>
|
|||
/// 不健康
|
|||
/// </summary>
|
|||
Unhealthy |
|||
} |
|||
|
|||
/// <summary>
|
|||
/// 处理器配置
|
|||
/// </summary>
|
|||
public class HandlerConfiguration |
|||
{ |
|||
/// <summary>
|
|||
/// 是否启用
|
|||
/// </summary>
|
|||
public bool IsEnabled { get; set; } = true; |
|||
|
|||
/// <summary>
|
|||
/// 超时时间（毫秒）
|
|||
/// </summary>
|
|||
public int Timeout { get; set; } = 30000; |
|||
|
|||
/// <summary>
|
|||
/// 最大重试次数
|
|||
/// </summary>
|
|||
public int MaxRetries { get; set; } = 3; |
|||
|
|||
/// <summary>
|
|||
/// 健康检查间隔（秒）
|
|||
/// </summary>
|
|||
public int HealthCheckInterval { get; set; } = 60; |
|||
|
|||
/// <summary>
|
|||
/// 最大失败次数阈值
|
|||
/// </summary>
|
|||
public int MaxFailureThreshold { get; set; } = 10; |
|||
|
|||
/// <summary>
|
|||
/// 平均处理时间警告阈值（毫秒）
|
|||
/// </summary>
|
|||
public double AverageProcessingTimeWarningThreshold { get; set; } = 1000; |
|||
} |
|||
} |
|||
@ -0,0 +1,79 @@ |
|||
using System; |
|||
using System.Threading.Tasks; |
|||
using CellularManagement.WebSocket.Models; |
|||
|
|||
namespace CellularManagement.WebSocket.Models |
|||
{ |
|||
/// <summary>
|
|||
/// 消息处理器接口
|
|||
/// </summary>
|
|||
public interface IMessageHandler : IDisposable |
|||
{ |
|||
/// <summary>
|
|||
/// 处理器优先级（数字越小优先级越高）
|
|||
/// </summary>
|
|||
int Priority { get; } |
|||
|
|||
/// <summary>
|
|||
/// 处理器名称
|
|||
/// </summary>
|
|||
string Name { get; } |
|||
|
|||
/// <summary>
|
|||
/// 处理消息
|
|||
/// </summary>
|
|||
/// <param name="message">WebSocket消息</param>
|
|||
/// <param name="cancellationToken">取消令牌</param>
|
|||
/// <returns>处理结果</returns>
|
|||
Task<WebSocketMessage> HandleAsync(WebSocketMessage message, CancellationToken cancellationToken); |
|||
|
|||
/// <summary>
|
|||
/// 获取处理器统计信息
|
|||
/// </summary>
|
|||
/// <returns>统计信息</returns>
|
|||
HandlerStatistics GetStatistics(); |
|||
|
|||
/// <summary>
|
|||
/// 获取处理器指标
|
|||
/// </summary>
|
|||
/// <returns>处理器指标</returns>
|
|||
HandlerMetrics GetMetrics(); |
|||
|
|||
/// <summary>
|
|||
/// 更新处理器配置
|
|||
/// </summary>
|
|||
/// <param name="configuration">处理器配置</param>
|
|||
void UpdateConfiguration(HandlerConfiguration configuration); |
|||
} |
|||
|
|||
/// <summary>
|
|||
/// 处理器统计信息
|
|||
/// </summary>
|
|||
public class HandlerStatistics |
|||
{ |
|||
/// <summary>
|
|||
/// 处理消息总数
|
|||
/// </summary>
|
|||
public long TotalMessages { get; set; } |
|||
|
|||
/// <summary>
|
|||
/// 成功处理消息数
|
|||
/// </summary>
|
|||
public long SuccessMessages { get; set; } |
|||
|
|||
/// <summary>
|
|||
/// 失败消息数
|
|||
/// </summary>
|
|||
public long FailedMessages { get; set; } |
|||
|
|||
/// <summary>
|
|||
/// 平均处理时间（毫秒）
|
|||
/// </summary>
|
|||
public double AverageProcessingTime { get; set; } |
|||
|
|||
/// <summary>
|
|||
/// 最后处理时间
|
|||
/// </summary>
|
|||
public DateTime LastProcessedTime { get; set; } |
|||
} |
|||
} |
|||
@ -0,0 +1,19 @@ |
|||
using System; |
|||
using System.Collections.Generic; |
|||
using System.Linq; |
|||
using System.Text; |
|||
using System.Threading.Tasks; |
|||
|
|||
namespace CellularManagement.WebSocket.Models |
|||
{ |
|||
/// <summary>
|
|||
/// 消息统计信息
|
|||
/// </summary>
|
|||
public class MessageStats |
|||
{ |
|||
public long TotalMessages { get; set; } |
|||
public long SuccessfulMessages { get; set; } |
|||
public long FailedMessages { get; set; } |
|||
public double AverageSendTime { get; set; } |
|||
} |
|||
} |
|||
@ -0,0 +1,152 @@ |
|||
using System.Text.Json; |
|||
using CellularManagement.WebSocket.Connection; |
|||
using CellularManagement.WebSocket.Models; |
|||
using Microsoft.Extensions.Logging; |
|||
using System.Collections.Concurrent; |
|||
using System.Diagnostics; |
|||
|
|||
namespace CellularManagement.WebSocket.Services; |
|||
|
|||
/// <summary>
|
|||
/// 出站消息处理器
|
|||
/// 负责处理消息的发送、重试、超时控制和监控
|
|||
/// </summary>
|
|||
public class OutgoingMessageProcessor : IDisposable |
|||
{ |
|||
private readonly WebSocketConnectionManager _connectionManager; |
|||
private readonly ILogger<OutgoingMessageProcessor> _logger; |
|||
private readonly ConcurrentDictionary<string, MessageStats> _messageStats; |
|||
private readonly SemaphoreSlim _backpressureSemaphore; |
|||
private readonly int _maxConcurrentSends; |
|||
private readonly int _maxRetryAttempts; |
|||
private readonly TimeSpan _sendTimeout; |
|||
private readonly TimeSpan _retryDelay; |
|||
private readonly CancellationTokenSource _stoppingCts; |
|||
|
|||
public OutgoingMessageProcessor( |
|||
WebSocketConnectionManager connectionManager, |
|||
ILogger<OutgoingMessageProcessor> logger, |
|||
int maxConcurrentSends = 100, |
|||
int maxRetryAttempts = 3, |
|||
int sendTimeoutSeconds = 30, |
|||
int retryDelaySeconds = 5) |
|||
{ |
|||
_connectionManager = connectionManager; |
|||
_logger = logger; |
|||
_maxConcurrentSends = maxConcurrentSends; |
|||
_maxRetryAttempts = maxRetryAttempts; |
|||
_sendTimeout = TimeSpan.FromSeconds(sendTimeoutSeconds); |
|||
_retryDelay = TimeSpan.FromSeconds(retryDelaySeconds); |
|||
_messageStats = new ConcurrentDictionary<string, MessageStats>(); |
|||
_backpressureSemaphore = new SemaphoreSlim(maxConcurrentSends); |
|||
_stoppingCts = new CancellationTokenSource(); |
|||
} |
|||
|
|||
/// <summary>
|
|||
/// 处理出站消息
|
|||
/// </summary>
|
|||
public async Task ProcessOutgoingMessagesAsync(CancellationToken stoppingToken) |
|||
{ |
|||
try |
|||
{ |
|||
await foreach (var message in _connectionManager.ReadOutgoingMessagesAsync(stoppingToken)) |
|||
{ |
|||
await ProcessMessageAsync(message, stoppingToken); |
|||
} |
|||
} |
|||
catch (OperationCanceledException) |
|||
{ |
|||
_logger.LogInformation("出站消息处理已停止"); |
|||
} |
|||
} |
|||
|
|||
private async Task ProcessMessageAsync(WebSocketMessage message, CancellationToken stoppingToken) |
|||
{ |
|||
await _backpressureSemaphore.WaitAsync(stoppingToken); |
|||
try |
|||
{ |
|||
var stats = _messageStats.GetOrAdd(message.ConnectionId, _ => new MessageStats()); |
|||
stats.TotalMessages++; |
|||
|
|||
var stopwatch = Stopwatch.StartNew(); |
|||
var success = await SendMessageWithRetryAsync(message, stoppingToken); |
|||
stopwatch.Stop(); |
|||
|
|||
if (success) |
|||
{ |
|||
stats.SuccessfulMessages++; |
|||
stats.AverageSendTime = (stats.AverageSendTime * (stats.SuccessfulMessages - 1) + stopwatch.ElapsedMilliseconds) / stats.SuccessfulMessages; |
|||
} |
|||
else |
|||
{ |
|||
stats.FailedMessages++; |
|||
} |
|||
} |
|||
finally |
|||
{ |
|||
_backpressureSemaphore.Release(); |
|||
} |
|||
} |
|||
|
|||
private async Task<bool> SendMessageWithRetryAsync(WebSocketMessage message, CancellationToken stoppingToken) |
|||
{ |
|||
for (int attempt = 1; attempt <= _maxRetryAttempts; attempt++) |
|||
{ |
|||
try |
|||
{ |
|||
using var cts = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken, _stoppingCts.Token); |
|||
cts.CancelAfter(_sendTimeout); |
|||
|
|||
var connection = _connectionManager.GetConnection(message.ConnectionId); |
|||
if (connection?.Socket.State != System.Net.WebSockets.WebSocketState.Open) |
|||
{ |
|||
_logger.LogWarning("连接不存在或已关闭，连接ID：{ConnectionId}", message.ConnectionId); |
|||
return false; |
|||
} |
|||
|
|||
await connection.Socket.SendAsync( |
|||
new ArraySegment<byte>(message.Data), |
|||
message.MessageType, |
|||
true, |
|||
cts.Token); |
|||
|
|||
_logger.LogDebug("消息发送成功，连接ID：{ConnectionId}，尝试次数：{Attempt}", |
|||
message.ConnectionId, attempt); |
|||
return true; |
|||
} |
|||
catch (OperationCanceledException) |
|||
{ |
|||
_logger.LogWarning("消息发送超时，连接ID：{ConnectionId}，尝试次数：{Attempt}", |
|||
message.ConnectionId, attempt); |
|||
} |
|||
catch (Exception ex) |
|||
{ |
|||
_logger.LogError(ex, "发送消息时发生错误，连接ID：{ConnectionId}，尝试次数：{Attempt}", |
|||
message.ConnectionId, attempt); |
|||
} |
|||
|
|||
if (attempt < _maxRetryAttempts) |
|||
{ |
|||
await Task.Delay(_retryDelay, stoppingToken); |
|||
} |
|||
} |
|||
|
|||
return false; |
|||
} |
|||
|
|||
/// <summary>
|
|||
/// 获取消息统计信息
|
|||
/// </summary>
|
|||
public IReadOnlyDictionary<string, MessageStats> GetMessageStats() |
|||
{ |
|||
return _messageStats; |
|||
} |
|||
|
|||
public void Dispose() |
|||
{ |
|||
_stoppingCts.Cancel(); |
|||
_backpressureSemaphore.Dispose(); |
|||
_stoppingCts.Dispose(); |
|||
} |
|||
} |
|||
|
|||
Loading…
Reference in new issue