如何在aio_pika中通过死信交换机拒绝消息并清除原队列消息？

核心原因 #

你代码里用了async with message.process(ignore_processed=True)上下文管理器，这个方法会自动对消息执行确认（ack）操作，哪怕你在上下文里调用reject，也会因为消息状态已经被标记为处理完成，导致reject不生效——死信队列能收到消息，但原队列的消息不会被移除，进而触发重复消费。

修复方案 #

直接去掉message.process()的上下文包裹，手动管理消息的确认/拒绝逻辑：

from aio_pika.abc import AbstractIncomingMessage
import json
import log

async def callback(
    message: AbstractIncomingMessage,
) -> None:
    try:
        parsed = json.loads(message.body)
        # 替换为你的业务判断逻辑
        success = False
        if not success:
            log.msg("[x] Reject")
            # 设置requeue=False，原队列消息会被移除，同时转发至死信队列（需提前配置死信参数）
            await message.reject(requeue=False)
        else:
            log.msg("[x] Updated!")
            # 处理成功时手动确认，原队列移除消息
            await message.ack()
    except Exception as e:
        # 异常场景也建议处理，避免消息长期卡队列
        log.msg(f"[x] Error occurred: {str(e)}")
        await message.reject(requeue=False)

关键注意点 #

• 确保目标队列已正确配置死信交换机（x-dead-letter-exchange）和死信路由键（x-dead-letter-routing-key），否则reject(requeue=False)会直接丢弃消息，不会转发到死信队列。
• 禁止混合使用message.process()和手动ack/reject，二者互斥会导致消息状态混乱。

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