如何配置RabbitMQ Fanout Exchange及队列，实现离线场景无消息丢失

核心配置思路 #

要保证消费者离线数分钟甚至数小时期间不丢失消息，核心是让Exchange、队列、消息都具备持久化特性，同时采用消费者手动确认机制，避免RabbitMQ提前删除未处理的消息。

具体配置步骤 #

1. 创建持久化的Fanout Exchange #

创建Exchange时必须指定durable: true，确保RabbitMQ重启后Exchange不会丢失。
示例代码（以Python pika为例）：

channel.exchange_declare(
    exchange='fanout_persistent',
    exchange_type='fanout',
    durable=True  # 开启Exchange持久化
)

2. 为每个消费者创建专属持久化队列 #

每个消费者需要独立队列（Fanout Exchange会将消息路由到所有绑定队列），队列必须设置durable: true，保证消费者离线或RabbitMQ重启时队列及其中的消息不会被删除。
示例代码：

# 消费者1的专属队列
channel.queue_declare(queue='consumer1_queue', durable=True)
# 消费者2的专属队列
channel.queue_declare(queue='consumer2_queue', durable=True)

3. 绑定队列到Fanout Exchange #

将两个持久化队列绑定到持久化的Fanout Exchange，绑定关系默认持久化，无需额外配置：

channel.queue_bind(exchange='fanout_persistent', queue='consumer1_queue')
channel.queue_bind(exchange='fanout_persistent', queue='consumer2_queue')

4. 生产者发送持久化消息 #

生产者发送消息时需设置delivery_mode=2，标记消息为持久化，确保消息被写入RabbitMQ磁盘存储，重启后不会丢失：

channel.basic_publish(
    exchange='fanout_persistent',
    routing_key='',  # Fanout Exchange无需指定路由键
    body='业务消息内容',
    properties=pika.BasicProperties(
        delivery_mode=2,  # 标记为持久化消息
    )
)

5. 消费者配置手动消息确认 #

消费者必须关闭自动确认，启用手动确认机制（auto_ack=False），只有当消息被成功处理后，才向RabbitMQ发送确认信号，避免消费者离线时消息被提前删除：

def callback(ch, method, properties, body):
    # 此处编写消息处理的业务逻辑
    print(f"消费者1处理消息: {body.decode()}")
    # 手动确认消息已处理完成
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_consume(
    queue='consumer1_queue',
    on_message_callback=callback,
    auto_ack=False  # 关闭自动确认
)

关键注意事项 #

• 三个持久化条件必须同时满足：Exchange持久化、队列持久化、消息持久化，缺一都会导致消息丢失风险。
• 手动确认机制是核心：若开启自动确认，RabbitMQ会在消息发送给消费者后立即删除，消费者离线时消息直接丢失。
• 不要给队列设置过短的TTL（存活时间）或过小的最大长度，避免离线期间消息被自动清理（默认无TTL和长度限制）。

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