MassTransit RabbitMQ 获取队列大小

Question

如果队列中的消息数量大于某个阈值，我希望生产者不要发送新消息。 为此，我需要找出队列中当前有多少消息。 有没有办法使用 MassTransit 框架来做到这一点？

Answer 1

MassTransit 仅使用交通工具的功能（您没有提到）。队列大小和其他测量通常是各种管理 API 的一部分。例如，可以在没有管理 API（这是一个插件）的情况下部署 RMQ 集群或实例，因此无论如何您都无法获取这些测量结果。

对于 RMQ，您可以使用管理 API 或包装器（例如 EasyNetQ 管理 API）来获取您需要的指标。

Answer 2

我知道这是一个老问题，但这就是我解决的方法，以防有人仍在寻找解决方案。

请注意，我在docker中运行了rabbitmq:management

public static class QueueInfo
{
    public async static Task<RabbitMqQueueInfo> GetQueueInfo(string queueName)
    {
        var httpClient = new HttpClient();
        var rabbitMqApiUrl = "http://localhost:15672/api";
        var rabbitMqVhost = "%2F";
        var rabbitMqApiUser = "guest";
        var rabbitMqApiPassword = "guest";
        var queueInfoUrl = $"{rabbitMqApiUrl}/queues/{rabbitMqVhost}/{queueName}";

        httpClient.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue(
            "Basic", Convert.ToBase64String(Encoding.ASCII.GetBytes($"{rabbitMqApiUser}:{rabbitMqApiPassword}")));

        var response = await httpClient.GetAsync(queueInfoUrl);

        if (!response.IsSuccessStatusCode)
        {
            throw new Exception($"Failed to retrieve queue information for queue '{queueName}'.");
        }

        var responseContent = await response.Content.ReadAsStringAsync();
        var queueInfo = JsonSerializer.Deserialize<RabbitMqQueueInfo>(responseContent);

        if (queueInfo == null)
        {
            throw new Exception($"Failed to deserialize queue information for queue '{queueName}'.");
        }

        return queueInfo;
    }

    public class RabbitMqQueueInfo
    {
        public int messages { get; set; }
        public QueueStatus backing_queue_status { get; set; }
    }

    public class QueueStatus
    {
        public double avg_ack_egress_rate { get; set; }
        public double avg_ack_ingress_rate { get; set; }
        public double avg_egress_rate { get; set; }
        public double avg_ingress_rate { get; set; }
    }
}