RabbitMQ：完整任务生产消费 CLI
本轮仅静态审阅，以下命令与练习均未执行；预期结果不是本轮运行证据。早期真实服务验证仅为历史记录，不能代替当前环境验证。
安装与配置
按 起点准备隔离服务后安装锁定依赖。下面 python 代表 Python 3.12；Windows 可用 py -3.12 替代。项目使用 --target 安装至本地 .venv 并自动加入模块查找路径，避免依赖全局环境。
txt
pika==1.3.2sh
python -m pip install --target examples/rabbitmq/.venv -r examples/rabbitmq/requirements.txt
python examples/rabbitmq/test_contract.pyPika 1.3.2 是 AMQP 0-9-1 驱动，无额外运行依赖，不能把它的版本号和 RabbitMQ 4.2.5 对齐比较。连接使用有限 socket_timeout、stack_timeout、blocked_connection_timeout；它们是退出边界，不是业务消息 TTL。
完整源码
代码包括连接、拓扑、持久发布、确认、mandatory、手动确认、nack、死信、空队列超时和顶层异常退出。它不创建 HTTP 后台或管理 API 工程。
python
"""AMQP 0-9-1 CLI with confirms, mandatory routing and explicit acknowledgements."""
import argparse
import json
import os
import pathlib
import sys
import time
sys.path.insert(0, os.environ.get('MESSAGING_PYTHON_DEPS', str(pathlib.Path(__file__).with_name('.venv'))))
import pika
def require_message(body):
if body is None:
raise TimeoutError('queue remained empty before deadline')
return body
def connect(host='127.0.0.1', port=25672):
return pika.BlockingConnection(pika.ConnectionParameters(
host, port, credentials=pika.PlainCredentials('student', 'local-lab-only'),
connection_attempts=1, socket_timeout=3, stack_timeout=6,
blocked_connection_timeout=5, heartbeat=30))
def topology(channel, prefix):
channel.exchange_declare(prefix, exchange_type='direct', durable=True)
channel.exchange_declare(prefix + '.dlx', exchange_type='direct', durable=True)
channel.queue_declare(prefix + '.dead', durable=True)
channel.queue_bind(prefix + '.dead', prefix + '.dlx', 'dead')
channel.queue_declare(prefix + '.work', durable=True, arguments={
'x-dead-letter-exchange': prefix + '.dlx', 'x-dead-letter-routing-key': 'dead'})
channel.queue_bind(prefix + '.work', prefix, 'created')
channel.basic_qos(prefetch_count=1)
def publish(channel, prefix, value, route='created'):
channel.confirm_delivery()
channel.basic_publish(prefix, route, value.encode(), mandatory=True,
properties=pika.BasicProperties(delivery_mode=2, content_type='text/plain',
message_id=value))
return {'confirmed': True, 'route': route, 'value': value}
def receive(channel, queue, action='ack', timeout=5, sink=None):
deadline, method, properties, body = time.monotonic() + timeout, None, None, None
while time.monotonic() < deadline:
method, properties, body = channel.basic_get(queue, auto_ack=False)
if method:
break
channel.connection.process_data_events(time_limit=.1)
require_message(body)
row = {'value': body.decode(), 'redelivered': method.redelivered,
'headers': properties.headers or {}, 'action': action}
# The CLI's effect is an explicitly flushed output, not an external business transaction.
if sink is not None:
sink(row)
if action == 'ack':
channel.basic_ack(method.delivery_tag)
elif action == 'requeue':
channel.basic_nack(method.delivery_tag, requeue=True)
elif action == 'dead':
channel.basic_nack(method.delivery_tag, requeue=False)
# leave deliberately does not ack. Closing the connection makes it eligible again.
return row
def main():
parser = argparse.ArgumentParser()
parser.add_argument('--host', default=os.environ.get('RABBITMQ_HOST', '127.0.0.1'))
parser.add_argument('--port', type=int, default=int(os.environ.get('RABBITMQ_PORT', '25672')))
parser.add_argument('--prefix', default='orders')
sub = parser.add_subparsers(dest='command', required=True)
sub.add_parser('setup')
send = sub.add_parser('produce')
send.add_argument('--value', required=True)
send.add_argument('--route', default='created')
read = sub.add_parser('consume')
read.add_argument('--action', choices=['ack', 'requeue', 'dead', 'leave'], default='ack')
read.add_argument('--dead', action='store_true')
read.add_argument('--timeout', type=float, default=5)
args = parser.parse_args()
connection = connect(args.host, args.port)
try:
channel = connection.channel()
topology(channel, args.prefix)
if args.command == 'produce':
result = publish(channel, args.prefix, args.value, args.route)
elif args.command == 'consume':
receive(channel, args.prefix + ('.dead' if args.dead else '.work'),
args.action, args.timeout,
sink=lambda row: print(json.dumps(row, ensure_ascii=False, default=str), flush=True))
return
else:
result = {'ready': args.prefix}
print(json.dumps(result, ensure_ascii=False, default=str))
finally:
connection.close()
if __name__ == '__main__':
try:
main()
except Exception as error:
print(f'{type(error).__name__}: {error}', file=sys.stderr)
sys.exit(1)从发布到领取
sh
python examples/rabbitmq/main.py --prefix course-jobs setup
python examples/rabbitmq/main.py --prefix course-jobs produce --value job-001
python examples/rabbitmq/main.py --prefix course-jobs consume --action ack
python examples/rabbitmq/main.py --prefix course-jobs consume --timeout 1
python examples/rabbitmq/main.py --port 1 --prefix course-jobs setup使用全新 course-jobs、队列原本为空且服务就绪时，前三条预期退出 0，发布 JSON 中 confirmed=true，消费 JSON 中 value=job-001；第一次投递通常 redelivered=false。第四条因队列为空而 TimeoutError，最后一条因连接拒绝而非零，不能用吞异常的空结果代替失败。
prefix 隔离一组 exchange/queue；测试使用随机 prefix，避免前次未确认或残留消息污染结果。不会自动 purge 现有数据。同名资源参数不匹配会明确失败，需要新 prefix 或经审阅的迁移流程。
历史验证器与完整工程文件
历史运行验证器为仓库 migration/knowledge-rebuild/m4-next/messaging/verify.py，本轮不执行。它会启动、停止和重启两种服务并改写结果文件，不属于静态检查命令。
该历史脚本设计为同时验证 Kafka，先按 Kafka 项目装好依赖。它观察实际消息值、redelivered 和 x-death，验证确认、不可路由、requeue、拒绝死信、未确认连接关闭重投、空队列错误、离线错误和受控重启保留。等待基于 AMQP 连接/channel 就绪、实际领取与 deadline，不靠 sleep 假定重投完成。逐命令退出和输出保存在 results.json。
下面补齐工程其余文件；Compose 已在 起步同源展示。契约测试会导入真实驱动；readiness 会发起实际 AMQP 连接并创建 channel，不声明资源或收发消息。这些文件均未在本轮执行。
python
"""Side-effect-free AMQP readiness: no declaration, publishing or consumption."""
from main import connect
connection = connect()
channel = connection.channel()
channel.close()
connection.close()
print('AMQP connection and channel ready')python
import importlib.util
import pathlib
import unittest
class Contract(unittest.TestCase):
def test_empty_delivery_is_not_success(self):
path = pathlib.Path(__file__).with_name('main.py')
spec = importlib.util.spec_from_file_location('lab', path)
lab = importlib.util.module_from_spec(spec)
spec.loader.exec_module(lab)
with self.assertRaises(TimeoutError):
lab.require_message(None)
self.assertEqual(lab.require_message(b'hello'), b'hello')
def test_sink_failure_does_not_ack(self):
path = pathlib.Path(__file__).with_name('main.py')
spec = importlib.util.spec_from_file_location('lab', path)
lab = importlib.util.module_from_spec(spec)
spec.loader.exec_module(lab)
class Method:
redelivered, delivery_tag = False, 1
class Properties:
headers = {}
class Channel:
acked = False
def basic_get(self, queue, auto_ack):
return Method(), Properties(), b'job'
def basic_ack(self, tag):
self.acked = True
channel = Channel()
def broken_sink(row):
raise OSError('output closed')
with self.assertRaises(OSError):
lab.receive(channel, 'queue', sink=broken_sink)
self.assertFalse(channel.acked)
if __name__ == '__main__':
unittest.main()md
# RabbitMQ 4.2.5 / Pika 1.3.2 laboratory
Static review only: no commands, tests, imports, installation or services below were executed. Earlier live results are historical evidence only.
Teaching procedure (not executed), from the repository root with Docker, Compose v2 and Python 3.12:
```sh
docker compose -p showdocs-m4b-rabbitmq -f examples/rabbitmq/compose.yaml up -d
python -m pip install --target examples/rabbitmq/.venv -r examples/rabbitmq/requirements.txt
python examples/rabbitmq/readiness.py
python examples/rabbitmq/test_contract.py
python examples/rabbitmq/main.py --prefix my-jobs setup
python examples/rabbitmq/main.py --prefix my-jobs produce --value job-1
python examples/rabbitmq/main.py --prefix my-jobs consume --action requeue
python examples/rabbitmq/main.py --prefix my-jobs consume --action dead
python examples/rabbitmq/main.py --prefix my-jobs consume --dead
```
The producer uses publisher confirms plus mandatory routing and persistent messages. Consumption flushes JSON before the requested ack/nack. `leave` deliberately closes without ack. A sink failure must not ack. These are bounded protocol demonstrations, not external business transactions; duplicates still need application deduplication. CLI errors return 1, argument errors return 2. Check the exit code even when output was already written.
AMQP: `127.0.0.1:25672`; optional management: `127.0.0.1:25673`; local-only account: `student` / `local-lab-only`. An init container fixes ownership of only this named volume and its cookie without reading cookie contents. The broker uses a stable hostname. Set `RABBITMQ_HOST`, `RABBITMQ_PORT` and `MESSAGING_PYTHON_DEPS` for a Linux client in the Compose network.
Install Kafka dependencies too and run `python migration/knowledge-rebuild/m4-next/messaging/verify.py` for the full suite; `pwsh -File migration/knowledge-rebuild/m4-next/messaging/verify-linux.ps1` runs ephemeral Linux clients. 课程正文入口：`docs/middleware/messaging/rabbitmq/plus/practice/project.md`。
`docker compose -p showdocs-m4b-rabbitmq -f examples/rabbitmq/compose.yaml down` keeps the data volume. Adding `-v` removes all laboratory queues/messages. This is a single-node classic-queue lab, not quorum HA, power-loss validation or an offsite backup.
就绪探测可能需要有限重试，约80秒仍失败时停止并查看日志。使用新的 prefix；已有队列消息、死信和未确认投递会改变领取顺序及结果。以上流程中，requeue 将 job-1 放回工作队列，dead 将其移入死信队列，最后 --dead 领取并确认。成功步骤预期退出0并显示对应消息；错误端点和有界空队列领取应非零退出，均不是本轮观察结果。历史迁移验证器会启停服务并写结果，本轮静态审读不执行它。--timeout 是重复轮询的期限，不是进程硬截止；连接、单次网络操作、提交与关闭仍可能另耗时间。CLI 的 value/body 合同是 UTF-8 文本，外部二进制记录不在合同内，解码失败不确认，关闭连接后可能重投。先按 练习自行预测结果，再核对答案。
继续阅读
- 知识专题
- 知识专题
- 知识专题