一文讲透RabbitMQ 消息队列中的拒绝机制

import pika

def callback(ch, method, properties, body):
    print(f"Received message: {body}")
    # 模拟处理失败
    if some_condition:
        ch.basic_reject(delivery_tag=method.delivery_tag, requeue=True)
    else:
        ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_consume(queue='my_queue', on_message_callback=callback)

def callback(ch, method, properties, body):
    print(f"Received message: {body}")
    # 模拟处理失败
    if some_condition:
        ch.basic_nack(delivery_tag=method.delivery_tag, multiple=False, requeue=True)
    else:
        ch.basic_ack(delivery_tag=method.delivery_tag)

channel.queue_declare(
    queue='my_queue',
    arguments={
        'x-dead-letter-exchange': 'dlx_exchange',  # 死信交换器名称
        'x-message-ttl': 60000  # 可选：消息过期时间（毫秒）
    }
)

channel.exchange_declare(exchange='dlx_exchange', exchange_type='direct')
channel.queue_declare(queue='dlx_queue')
channel.queue_bind(exchange='dlx_exchange', queue='dlx_queue', routing_key='dlx_key')

def callback(ch, method, properties, body):
    if some_condition:
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
    else:
        ch.basic_ack(delivery_tag=method.delivery_tag)

@RabbitListener(queues = "my_queue", ackMode = "MANUAL")
public void onMessage(Message message, Channel channel) throws IOException {
    try {
        // 处理消息
        if (someCondition) {
            // 拒绝消息并重新入队
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
        } else {
            // 确认消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }
    } catch (Exception e) {
        // 异常处理
        channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
    }
}

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: auto

public class MyConsumer implements MessageListenerConcurrently {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        try {
            // 处理消息
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        } catch (Exception e) {
            // 消费失败，重新投递
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }
}

consumer.setMaxReconsumeTimes(3); // 最大重试3次