.NET Core WebAPI对接RabbitMQ时出现随机消息丢失问题求助

问题根因 #

• 接收端连接/通道生命周期错误：你使用using块包裹了RabbitMQ连接和通道，Receive方法执行完成后会立刻销毁连接和通道资源。EventingBasicConsumer的Received事件是异步触发的，绝大多数情况下回调还未执行，连接就已经断开，消息直接丢失。
• 自动确认配置不合理：BasicConsume的第二个参数设为true开启了自动确认，RabbitMQ只要将消息推送给消费者就会直接将消息从队列删除，不管消费者是否真的完成了消息处理，所以你会观察到队列清空但控制台没有输出的现象。
• 消费模式错误：RabbitMQ消费者需要保持长连接运行，不应该作为接口触发的短生命周期逻辑，每次调用接口才临时启动消费者的模式本身不符合RabbitMQ的使用规范。

修复方案 #

1. 接收端改用后台长驻消费者 #

在ASP.NET Core中使用BackgroundService实现长时间运行的消费者，避免连接被提前释放：

public class RabbitMqConsumerService : BackgroundService
{
    private readonly string _queueName = "test-queue";
    private IConnection _connection;
    private IModel _channel;

    protected override Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var factory = new ConnectionFactory() { HostName = "localhost" };
        _connection = factory.CreateConnection();
        _channel = _connection.CreateModel();
        
        _channel.QueueDeclare(_queueName,
            durable: false,
            exclusive: false,
            autoDelete: false,
            arguments: null);

        var consumer = new EventingBasicConsumer(_channel);
        consumer.Received += (model, ea) =>
        {
            var msg = ea.Body.ToArray();
            var message = Encoding.UTF8.GetString(msg);
            Console.WriteLine(message);
            // 处理完成后手动确认消息
            _channel.BasicAck(ea.DeliveryTag, false);
        };
        // 关闭自动确认，改为手动确认
        _channel.BasicConsume(_queueName, false, consumer);

        stoppingToken.Register(() =>
        {
            _channel.Close();
            _connection.Close();
        });

        return Task.CompletedTask;
    }
}

然后在Program.cs中注册这个后台服务：

builder.Services.AddHostedService<RabbitMqConsumerService>();

注册完成后服务会随应用启动自动运行，保持长连接消费消息，不需要再写receive接口触发消费。

2. 发送端优化（可选） #

避免每次发送消息都新建连接，可复用单例连接和通道提升性能：

public static class QueueManager
{
    public static string queueName = "test-queue";
    public static int count = 0;
    private static IConnection _connection;
    private static IModel _channel;

    static QueueManager()
    {
        var factory = new ConnectionFactory() { HostName = "localhost" };
        _connection = factory.CreateConnection();
        _channel = _connection.CreateModel();
        _channel.QueueDeclare(queueName,
            durable: false,
            exclusive: false,
            autoDelete: false,
            arguments: null);
    }

    public static void Send()
    {
        count++;
        var message = new { Message = "Sent Message", count = count };
        var body = JsonSerializer.Serialize(message);
        var queueMessage = Encoding.UTF8.GetBytes(body);
        _channel.BasicPublish("", queueName, null, queueMessage);
    }
}

内容的提问来源于stack exchange，提问作者user16230372