using Fake; using Fake.EventBus.RabbitMQ; using Fake.RabbitMQ; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using Oms.Application.Dtos.Orders; using RabbitMQ.Client; namespace Oms.Application.Messaging; public interface IOrderBatchQueueInspector { bool IsEnabled { get; } Task GetStatsAsync(CancellationToken cancellationToken = default); } /// /// 巡检 Fake.EventBus.RabbitMQ 的工作队列与死信队列。 /// public sealed class OrderBatchQueueInspector( IRabbitMqConnectionPool connectionPool, IOptions options, IOptions eventBusOptions, IApplicationInfo applicationInfo, ILogger logger) : IOrderBatchQueueInspector { /// /// DLX 堆积告警默认阈值。 /// public const uint DefaultDlxAlertThreshold = 100; public bool IsEnabled => options.Value.Enabled; public Task GetStatsAsync(CancellationToken cancellationToken = default) { if (!options.Value.Enabled) { return Task.FromResult(new OrderBatchDlxStatsOutput { Enabled = false, DlxAlertThreshold = DefaultDlxAlertThreshold, }); } var busOptions = eventBusOptions.Value; // 与 RabbitMqEventBus 一致:QueueName 为空时用 ApplicationName var queueName = (busOptions.QueueName ?? applicationInfo.ApplicationName).TrimEnd('.') + ".Queue"; var dlxQueueName = "DLX." + queueName; // Connection 由 Fake 连接池托管,禁止 Dispose var connection = connectionPool.Get(busOptions.ConnectionName); using var channel = connection.CreateModel(); var queues = new List { InspectQueue(channel, queueName), InspectQueue(channel, dlxQueueName), }; var dlxCount = queues.First(x => x.QueueName == dlxQueueName).MessageCount; logger.LogDebug("Order batch Fake EventBus queue stats, dlxMessages={DeadLetterCount}", dlxCount); return Task.FromResult(new OrderBatchDlxStatsOutput { Enabled = true, DeadLetterMessageCount = dlxCount, DlxAlertThreshold = DefaultDlxAlertThreshold, Alert = dlxCount >= DefaultDlxAlertThreshold, Queues = queues, }); } private static OrderBatchQueueStatOutput InspectQueue(IModel channel, string queueName) { try { var declare = channel.QueueDeclarePassive(queueName); return new OrderBatchQueueStatOutput { QueueName = queueName, MessageCount = declare.MessageCount, ConsumerCount = declare.ConsumerCount, }; } catch (Exception) { return new OrderBatchQueueStatOutput { QueueName = queueName }; } } }