RabbitMQ在C#中实现消息重新入队与重复处理的问题咨询

你遇到的问题根源是：使用BasicReject(deliveryTag, true)时，RabbitMQ会把被拒绝的消息放回队列头部，导致消费者会优先重复消费这条消息，后续消息（比如5）完全被阻塞，只能一直循环处理消息4。

以下是几种直接有效的解决方案：

1. 将消息放回队列尾部而非头部 #

用BasicNack拒绝当前消息，再手动将消息重新发布到队列尾部，这样消费者会先处理后续消息，之后再回来重试这条失败的消息。

修改后的代码：

consumer.Received += (sender, args) => {
    var body = args.Body.ToArray();
    var message = Encoding.UTF8.GetString(body);

    Console.WriteLine($"Message received by receiver1: {message}");

    if (Convert.ToInt32(message) == 4)
    {
        // 先拒绝当前消息，不自动重新入队
        channel.BasicNack(args.DeliveryTag, false, false);
        // 复制原消息属性，重新发布到队列尾部
        var properties = channel.CreateBasicProperties();
        properties.Persistent = args.BasicProperties.Persistent; // 保留原消息的持久化特性
        channel.BasicPublish(args.Exchange, args.RoutingKey, properties, body);
    }
    else
    {
        channel.BasicAck(args.DeliveryTag, false);
    }
};

2. 引入死信队列+重试次数限制（推荐方案） #

通过死信队列（DLQ）实现有限次数的重试，避免无限循环，同时不会阻塞正常消息的处理：

初始化队列配置（消费者启动时执行） #

// 声明死信交换机和死信队列
channel.ExchangeDeclare("main-dlx", ExchangeType.Direct, durable: true);
channel.QueueDeclare("main-dlq", durable: true, exclusive: false, autoDelete: false);
channel.QueueBind("main-dlq", "main-dlx", "main-routing-key");

// 声明主队列，绑定死信规则和重试延迟
var mainQueueArgs = new Dictionary<string, object>
{
    {"x-dead-letter-exchange", "main-dlx"},
    {"x-dead-letter-routing-key", "main-routing-key"},
    {"x-message-ttl", 5000} // 失败后5秒自动重试
};
channel.QueueDeclare("main-queue", durable: true, exclusive: false, autoDelete: false, arguments: mainQueueArgs);

// 设置消费者预取计数为1，确保一次只处理一条消息
channel.BasicQos(0, 1, false);

消费者处理逻辑 #

consumer.Received += (sender, args) => {
    var body = args.Body.ToArray();
    var message = Encoding.UTF8.GetString(body);
    Console.WriteLine($"Message received by receiver1: {message}");

    // 获取当前重试次数，默认0
    int retryCount = args.BasicProperties.Headers?.ContainsKey("retry-count") == true 
        ? (int)args.BasicProperties.Headers["retry-count"] 
        : 0;

    if (Convert.ToInt32(message) == 4)
    {
        retryCount++;
        if (retryCount > 3) // 重试3次后转入死信队列
        {
            channel.BasicReject(args.DeliveryTag, false);
            Console.WriteLine($"Message {message} failed after 3 retries, sent to DLQ");
        }
        else
        {
            // 更新重试次数，重新发布消息
            var properties = channel.CreateBasicProperties();
            properties.Headers = new Dictionary<string, object> {{"retry-count", retryCount}};
            properties.Persistent = args.BasicProperties.Persistent;
            
            // 拒绝当前消息，不自动入队
            channel.BasicReject(args.DeliveryTag, false);
            // 发布到主队列，利用TTL实现延迟重试
            channel.BasicPublish(args.Exchange, args.RoutingKey, properties, body);
            Console.WriteLine($"Message {message} retried {retryCount} times, requeued after delay");
        }
    }
    else
    {
        channel.BasicAck(args.DeliveryTag, false);
    }
};

这种方案既能保证消息不会丢失，又能避免无限循环阻塞队列，同时通过延迟重试给故障（比如数据库错误）留恢复时间。

3. 调整消费者预取计数 #

如果你的消费者设置了较高的预取计数，可能会一次性获取多条消息，导致卡住的消息阻塞其他已获取的消息。可以设置预取计数为1：

channel.BasicQos(0, 1, false);

这是辅助优化，核心还是要解决消息重入队的位置或重试次数限制问题。

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