3、RabbitMQ工作模式（一）：简单队列模式（Hello World）、Work Queues（工作队列）模式、消息确认机制
好，咱们正式开始聊 RabbitMQ 的具体用法。
说实话，很多初学者一上来就被各种交换机、路由键搞懵了。我个人习惯是，先别管那些花里胡哨的，从最简单的模式入手。今天我们就先啃下三块硬骨头：简单队列模式、工作队列模式，以及那个救命的消息确认机制。
3.1 简单队列模式（Hello World）
这个模式，说白了就是「一个生产者，一个消费者」。消息从生产者发出，经过一个队列，被消费者拿走。就像你给一个人发微信，他收到并看了。
我在项目中第一次用 RabbitMQ 就是这种模式。当时要做一个日志收集的 demo，简单到不行，但跑通的那一刻还是挺爽的。
3.1.1 核心流程
- 生产者连接到 RabbitMQ 服务器
- 生产者声明一个队列
- 生产者发送消息到该队列
- 消费者连接到同一个队列
- 消费者接收并处理消息
3.1.2 代码示例
先看生产者（send.py）：
import pika
# 建立连接
connection = pika.BlockingConnection(
pika.ConnectionParameters('localhost')
)
channel = connection.channel()
# 声明队列。如果队列不存在，就创建它
channel.queue_declare(queue='hello')
# 发送消息
channel.basic_publish(
exchange='',
routing_key='hello',
body='Hello World!'
)
print(" [x] 发送了 'Hello World!'")
# 关闭连接
connection.close()再看消费者（receive.py）：
import pika
connection = pika.BlockingConnection(
pika.ConnectionParameters('localhost')
)
channel = connection.channel()
# 同样声明队列。防止消费者先启动时队列不存在
channel.queue_declare(queue='hello')
# 定义回调函数，处理收到的消息
def callback(ch, method, properties, body):
print(f" [x] 收到消息: {body}")
# 告诉 RabbitMQ，这个回调函数用来处理队列中的消息
channel.basic_consume(
queue='hello',
on_message_callback=callback,
auto_ack=True
)
print(' [*] 等待消息。按 Ctrl+C 退出')
channel.start_consuming()3.2 Work Queues（工作队列）模式
简单队列只能一对一。但实际项目中，消息量一大，一个消费者根本忙不过来。这时候就需要工作队列模式了。
工作队列，也叫任务队列。它的核心思想是：一个生产者，多个消费者，共同消费同一个队列中的消息。每条消息只会被一个消费者消费。
你想想看，这就像你有一个任务列表，几个同事各自认领任务。谁有空谁干，干完一个再领下一个。
3.2.1 轮询分发（Round-robin dispatching）
默认情况下，RabbitMQ 会按顺序把消息轮流发给每个消费者。第一个消息给消费者A，第二个给消费者B，第三个又给A，以此类推。这叫轮询分发。
嗯，这里要注意：轮询分发不考虑消费者的处理能力。如果A处理慢，B处理快，A那边会积压，B却闲着。这显然不合理。
3.2.2 公平分发（Fair dispatch）
为了解决上面的问题，RabbitMQ 提供了公平分发机制。它通过 basic_qos(prefetch_count=1) 来实现。
这个设置的意思是：每次只给消费者发一条消息，等消费者处理完并确认后，再发下一条。这样，处理快的消费者就能多干活，处理慢的少干活。
3.2.3 代码示例：工作队列的生产者
import pika
import sys
connection = pika.BlockingConnection(
pika.ConnectionParameters('localhost')
)
channel = connection.channel()
channel.queue_declare(queue='task_queue', durable=True)
message = ' '.join(sys.argv[1:]) or "Hello World!"
channel.basic_publish(
exchange='',
routing_key='task_queue',
body=message,
properties=pika.BasicProperties(
delivery_mode=2, # 消息持久化
)
)
print(f" [x] 发送了 {message}")
connection.close()3.2.4 代码示例：工作队列的消费者（带公平分发）
import pika
import time
connection = pika.BlockingConnection(
pika.ConnectionParameters('localhost')
)
channel = connection.channel()
channel.queue_declare(queue='task_queue', durable=True)
# 关键：每次只取一条消息
channel.basic_qos(prefetch_count=1)
def callback(ch, method, properties, body):
print(f" [x] 收到 {body.decode()}")
# 模拟耗时任务
time.sleep(body.count(b'.'))
print(" [x] 处理完成")
# 手动发送确认
ch.basic_ack(delivery_tag=method.delivery_tag)
channel.basic_consume(
queue='task_queue',
on_message_callback=callback
)
print(' [*] 等待消息。按 Ctrl+C 退出')
channel.start_consuming()3.3 消息确认机制
这是 RabbitMQ 里最容易被忽视，但也是最关键的部分。没有它，你的系统随时可能丢消息。
我曾经在一个日志处理系统里踩过这个坑。消费者处理到一半崩溃了，消息没了，日志丢了。排查了半天才发现是自动确认惹的祸。
3.3.1 自动确认 vs 手动确认
| 确认方式 | 行为 | 风险 |
|---|---|---|
| 自动确认（auto_ack=True） | RabbitMQ 一发送消息，就认为消费成功 | 消费者崩溃，消息丢失 |
| 手动确认（auto_ack=False） | 消费者处理完，主动调用 basic_ack 通知 RabbitMQ | 不会丢消息，但需要正确处理 |
3.3.2 手动确认的正确姿势
- 消费者设置
auto_ack=False - 处理完业务逻辑后，调用
ch.basic_ack(delivery_tag=method.delivery_tag) - 如果处理失败，可以调用
basic_nack或basic_reject让消息重新入队
3.3.3 消息拒绝与重新入队
如果消费者发现消息有问题，比如格式错误，可以拒绝它：
# 拒绝单条消息，并重新入队
ch.basic_reject(delivery_tag=method.delivery_tag, requeue=True)
# 或者批量拒绝
ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True, multiple=False)注意：requeue=True 会让消息重新回到队列头部，可能被同一个消费者再次消费，导致死循环。我建议在拒绝时记录日志，并设置最大重试次数。
3.3.4 消息持久化
光有确认还不够。如果 RabbitMQ 服务器宕机了，内存里的消息全没了。怎么办？持久化。
持久化需要做两件事：
- 队列持久化：声明队列时设置
durable=True - 消息持久化：发送消息时设置
delivery_mode=2
但要注意，持久化会影响性能。如果对消息可靠性要求极高，可以持久化；如果允许少量丢失，用内存队列更快。
3.4 小结
今天咱们聊了三个核心概念：
- 简单队列模式：一对一，入门必备
- 工作队列模式：一对多，轮询或公平分发
- 消息确认机制：手动确认+持久化，保证消息不丢
说实话，这些基础模式虽然简单，但它们是后面所有高级模式（发布订阅、路由、主题）的基石。把今天的内容吃透了，后面的路就好走了。
下一章，咱们聊聊 RabbitMQ 的交换机，那才是真正灵活的地方。