企业官网建设流程全解析

1. 项目背景

支付要先问库存服务「能否预占 2 件 SKU」。有人用同步 HTTP，库存抖动时收银台一起抖。有人用 MQ 但只发 body：库存回了三条响应，订单服务把别人的预占结果安在当前用户头上，超卖事故。还有人把订单号放在 JSON 里却不设message_id，重试时库存按「新请求」再扣一次。

属性不是可选彩头，而是双方必须遵守的信封：

delivery_mode 持久化（第 7 章） content_type json / 二进制，避免乱解码 message_id 这条消息是谁，重试仍是它 correlation_id 这次 RPC 对话的配对号 reply_to 响应打回哪里 timestamp 超时与审计 expiration 单条 TTL 字符串毫秒 priority 需队列支持 x-max-priority headers 业务扩展，如 x-retry、tenant

RPC 在 MQ 上的经典做法：请求发到q.stock.rpc，reply_to指向调用方的exclusive 回调队列（第 7 章：人走格子拆），库存服务把同一correlation_id写回。超时则本地取消等待，迟到的响应当孤儿丢掉。乱序时只认匹配的 id，绝不按「先到先用」。

本章仍用手动拓扑与 pika。不要把 RPC 当普通通知：通知可以丢重试，预占必须配对、超时、幂等。

2. 项目设计

小胖把餐厅取餐小票举起来。

小胖：这不就是叫号吗？我报手机尾号，厨房喊号。为啥还要 correlation_id、reply_to 这么多栏？直接把库存结果发回订单队列不就行了，反正都是 MQ。

大师：叫号必须核对小票号，否则你端走邻桌的菜。订单队列是「通知广播」，人人都能听；RPC 响应必须回到这一次等待的人。reply_to 是你的桌号，correlation_id 是这一餐的小票。厨房不看小票只按到达顺序上菜，就会超卖或把失败结果给成功的人。

技术映射：reply_to = 回调队列；correlation_id = 配对键；message_id = 请求本身的身份（幂等）。

小白：回调队列用 exclusive 还是每个调用方固定q.order.reply？固定队列怎么避免别人收到我的响应？超时后响应才到会不会又扣一次库存？直接 reply-to 不声明队列行不行？属性里 expiration 和队列 TTL 谁优先？幂等键放 message_id 还是 headers 还是 body？AMQP 0-9-1 的 RPC 和 gRPC 比对吗？

大师：演示用 exclusive，进程内并发用 correlation_id 区分多次等待。固定共享回复队列要靠 id 过滤，还要防止别的服务误消费，权限上只给库存 write 该队列。超时后库存仍可能执行成功——必须库存侧也用 message_id 幂等，订单侧丢弃孤儿响应，必要时查库存状态对账，不能再发一条「看起来一样」的新 message_id。reply_to 写成队列名，对方basic.publish到默认交换机 + 该 routing key；未声明的 exclusive 名由 Broker 生成后填进去。expiration 与队列 TTL 取更短（第 10 章）。幂等键：message_id 表示「这一封信」，业务单号可同时放 headersx-order-id便于检索；只放 body 时重试网关可能改 JSON 空白导致哈希变。gRPC 适合同步强依赖；MQ RPC 适合解耦与削峰，但多了超时与孤儿，不要拿 MQ 模拟本地函数调用的所有语义。

小胖：那我一次预占就开一条 exclusive 队列，3000 QPS 会不会把 Broker 队列数打爆？

大师：会。高并发 RPC 应每进程一条回调队列（连接级），用 correlation_id 多路复用，而不是每次请求 declare 新队列。这和第 4 章「连接贵、通道便宜」是同一哲学：队列声明也贵。实验可以每次请求一条，好懂；生产规范写死进程级复用。

技术映射：回调队列数 ≈ 客户端进程数，不是请求数。

小白：乱序实验怎么做？库存能否在未声明 reply_to 时丢弃？priority 不配 x-max-priority 有用吗？content_encoding 要不要强制 utf-8？

大师：服务端故意 sleep 短请求更久、先回后到的请求，客户端必须仍匹配。无 reply_to 则库存 Ack 并打错误指标，不要往猜测的队列发。priority 没有队列参数就是空操作。content_type 用application/json，编码 utf-8 写进规范即可。

小胖：实验三枪：配对正确、乱序不串单、超时丢孤儿。属性表贴进 Wiki。

3. 项目实战

3.1 环境准备

使用第 11 章orderVHost 与app_order（若尚未 provision，用promo也可，但正文按隔离后的账号写）。库存服务可与订单同用户演示，生产应拆app_stock只授 RPC 队列。

# 若沿用 ch11exportVH=orderexportUSER=app_orderexportPASS=ord_dev_2026

3.2 步骤一：声明 RPC 工作队列

步骤目标：持久工作队列；回调队列由客户端 exclusive 声明。

# promo-mq/ch12/declare_rpc.pyimportpika conn=pika.BlockingConnection(pika.ConnectionParameters("127.0.0.1",5672,"order",pika.PlainCredentials("app_order","ord_dev_2026")))ch=conn.channel()ch.queue_declare("q.stock.rpc",durable=True)print("rpc work queue ready")conn.close()

3.3 步骤二：库存 worker（回 echo + 故意乱序）

步骤目标：按 correlation_id 回；对slowSKU sleep，制造乱序。

# promo-mq/ch12/stock_worker.pyimportjson,time,pika conn=pika.BlockingConnection(pika.ConnectionParameters("127.0.0.1",5672,"order",pika.PlainCredentials("app_order","ord_dev_2026"),heartbeat=30))ch=conn.channel()ch.queue_declare("q.stock.rpc",durable=True)ch.basic_qos(prefetch_count=4)defon_req(ch,method,props,body):req=json.loads(body)print("req",req,"cid",props.correlation_id,"reply",props.reply_to)ifreq.get("sku")=="SLOW":time.sleep(1.5)ok=req.get("qty",0)<=10resp=json.dumps({"ok":ok,"sku":req["sku"],"message_id":props.message_id})ifnotprops.reply_toornotprops.correlation_id:print("drop malformed rpc")ch.basic_ack(method.delivery_tag)returnch.basic_publish("",props.reply_to,resp.encode(),properties=pika.BasicProperties(correlation_id=props.correlation_id,content_type="application/json",delivery_mode=1,# 回调队列短命，可不 persistent),)ch.basic_ack(method.delivery_tag)ch.basic_consume("q.stock.rpc",on_req,auto_ack=False)print("stock worker up")ch.start_consuming()

坑：worker 里sleep会挡心跳，实验 1.5s 可接受；生产用线程池。
坑：向已消失的 exclusive 队列 publish 会 Return，要开 mandatory 或忽略。

3.4 步骤三：客户端配对、乱序、超时

步骤目标：并发两条（FAST 与 SLOW），SLOW 先发后到；再测超时。

# promo-mq/ch12/stock_client.pyimportjson,time,uuid,pikafrompika.exceptionsimportUnroutableErrorclassStockRpc:def__init__(self):self.conn=pika.BlockingConnection(pika.ConnectionParameters("127.0.0.1",5672,"order",pika.PlainCredentials("app_order","ord_dev_2026"),heartbeat=30))self.ch=self.conn.channel()self.ch.queue_declare("q.stock.rpc",durable=True)q=self.ch.queue_declare("",exclusive=True)self.callback=q.method.queue self.pending={}self.ch.basic_consume(self.callback,self._on_reply,auto_ack=True)def_on_reply(self,ch,method,props,body):fut=self.pending.pop(props.correlation_id,None)iffutisNone:print("orphan",props.correlation_id,body)returnfut["data"]=json.loads(body)fut["done"]=Truedefcall(self,sku,qty,timeout=2.0):cid=str(uuid.uuid4())mid=str(uuid.uuid4())self.pending[cid]={"done":False,"data":None}props=pika.BasicProperties(reply_to=self.callback,correlation_id=cid,message_id=mid,content_type="application/json",delivery_mode=2,timestamp=int(time.time()),headers={"x-order-id":"P-RPC","x-sku":sku},)self.ch.confirm_delivery()self.ch.basic_publish("","q.stock.rpc",json.dumps({"sku":sku,"qty":qty}).encode(),properties=props,mandatory=True)t0=time.time()whiletime.time()-t0<timeout:self.conn.process_data_events(time_limit=0.1)ifself.pending[cid]["done"]:returnself.pending.pop(cid)["data"]self.pending.pop(cid,None)# 超时后迟到的视为孤儿raiseTimeoutError(f"rpc timeout sku={sku}cid={cid}")if__name__=="__main__":rpc=StockRpc()# 先发慢的再发快的，完成顺序应是 FAST 先返回，但不能把 FAST 结果当成 SLOWimportthreading box={}defgo(name,sku):try:box[name]=rpc.call(sku,1,timeout=3.0)exceptExceptionase:box[name]=str(e)t1=threading.Thread(target=go,args=("slow","SLOW"))t2=threading.Thread(target=go,args=("fast","FAST"))t1.start();time.sleep(0.05);t2.start()t1.join();t2.join()print("box",box)# 注意：BlockingConnection 非线程安全，生产用独立连接或串行 wait

说明：pikaBlockingConnection不要多线程共享（第 4 章）。上面 threading 仅示意乱序；实验请串行：先 publish SLOW 与 FAST 两个 cid，在单线程process_data_events里看谁先done，断言slow的 data.sku 仍是 SLOW。

更稳的串行乱序脚本：

# promo-mq/ch12/rpc_reorder.pyimportjson,time,uuid,pika conn=pika.BlockingConnection(pika.ConnectionParameters("127.0.0.1",5672,"order",pika.PlainCredentials("app_order","ord_dev_2026")))ch=conn.channel()cb=ch.queue_declare("",exclusive=True).method.queue got={}defon_reply(ch,method,props,body):got[props.correlation_id]=json.loads(body)ch.basic_consume(cb,on_reply,auto_ack=True)ids={}forskuin("SLOW","FAST"):cid=str(uuid.uuid4())ids[sku]=cid ch.basic_publish("","q.stock.rpc",json.dumps({"sku":sku,"qty":1}).encode(),properties=pika.BasicProperties(reply_to=cb,correlation_id=cid,message_id=str(uuid.uuid4()),content_type="application/json",delivery_mode=2))t0=time.time()whiletime.time()-t0<4andlen(got)<2:conn.process_data_events(time_limit=0.2)print("FAST maps",got[ids["FAST"]]["sku"],"SLOW maps",got[ids["SLOW"]]["sku"])assertgot[ids["FAST"]]["sku"]=="FAST"assertgot[ids["SLOW"]]["sku"]=="SLOW"print("reorder OK")conn.close()

运行结果：即使 FAST 响应先到，字典仍按 cid 入座，不会串单。超时实验把 worker 关掉，call(..., timeout=0.5)应TimeoutError，随后启动 worker 打来的包打印orphan。

坑：超时后务必从pending删除，否则迟到包会填进已放弃的请求。
坑：message_id每次重试若换新，库存会当成新预占——重试应沿用同一 message_id。
坑：exclusive 回调队列不要 durable，也不可多连接共享。

属性在rabbit_basic里组装进内容帧，不是 JSON 的一部分：

-export([message/3, ... properties/1, extract_headers/1, extract_timestamp/1,

3.5 步骤四：属性验收表（测试用）

发一条通知（非 RPC）检查信封：

props=pika.BasicProperties(delivery_mode=2,content_type="application/json",message_id="mid-1",timestamp=int(time.time()),headers={"x-order-id":"P1"},expiration="60000")

HTTP Get 应能看到这些字段。缺content_type的发布在代码评审打回。

3.6 完整代码清单

column/samples/ch12/declare_rpc.py column/samples/ch12/stock_worker.py column/samples/ch12/rpc_reorder.py

3.7 测试验证

值班检查单：RPC 工作队列必须有消费者告警；回调队列数异常上涨说明有人按请求 declare。超时率进仪表盘。库存幂等表按 message_id 唯一。禁止把 RPC 响应发到 Fanout。文档写明「MQ RPC 不是本地调用」，产品不要假设一定 50ms 内返回。

单线程 BlockingConnection 做 RPC 时，消费回调与等待循环必须在同一线程跑process_data_events，否则看起来像「MQ 丢响应」，其实是客户端没泵事件。这和第 9 章 sleep 挡心跳是亲戚问题。

信封规范建议打印成一页纸：必填 message_id、content_type、delivery_mode；RPC 另加 correlation_id 与 reply_to；重试不得更换 message_id；headers 里的业务键只增不改名。代码评审对照这一页，比看 JSON schema 更快。库存预占的成功响应也要带原 message_id，便于订单侧日志对上。超时后的补偿不是再发一条全新 id，而是查询库存状态或带着旧 id 重试。产品若要求「用户连点两次必须扣两次」，那不是幂等场景，应拆成两笔预占而不是共享 id。

RPC 工作队列的消费者数告警与通知队列不同：RPC 没人接会导致收银台线程/协程堆超时，比短信晚到更伤转化。值班把q.stock.rpc的 consumers=0 定为 P1，短信队列可为 P2。不要用同一套「堆积深度」阈值套两种队列：RPC 深度一小就该响，因为它表示正在等待的用户。回调队列数突然升高，去查是否有人按请求 declare exclusive。Direct Reply-to 能缓解队列数，但排障更不直观，团队不熟就先 exclusive 复用。

乱序测试必须自动化断言 sku 字段，禁止用肉眼看打印顺序。CI 里 worker 与 client 的启动竞争要用就绪探针（队列存在且 consumers≥1）再发请求，否则用例红在「没 worker」而不是乱序逻辑。超时用例要关 worker 而不是把 timeout 设得比 sleep 还长还期望失败——那种用例测的是网络而不是客户端 pending 清理。孤儿响应打 metrics，不要只 print。

属性与 body 的职责切分：能放信封的不要只放 JSON，否则多语言客户端解码失败时连配对键都拿不到。correlation_id 尤其必须在属性上。body 可以缺字段，信封不能缺。审计系统若要检索，可冗余一份到 headers，但主键仍是属性。RPC 超时时间应小于用户接口超时，才能在 HTTP 层返回明确「预占未确认」而不是整串 504。超时后页面提示「请刷新订单状态」，后台对账，而不是让用户连点造成新的 message_id。把产品文案和 RPC 超时一起评审，否则技术做对了客服仍教用户狂点。

库存 worker 的 prefetch 不宜过大，预占是重操作。prefetch=4 的示例是起点，按 DB 耗时调。与第 9 章相同：窗口过大只是把等待从队列挪到 unacked。RPC 的 unacked 告警应比通知更敏感。worker 进程滚动时未 Ack 会回流，新实例必须幂等，这和第 9 章滚动双发是同一类事故，只是伤的是库存不是短信。

客户端 pending 字典要有上限，防止超时路径漏删导致内存涨。定期扫描超时条目。这是应用层的 connection_max，与 Broker 无关，但故障看起来像 MQ 泄漏。排障时同时看两边。

4. 项目总结

优点与缺点

优点：1）信封标准化可测。2）乱序可证明。3）超时与孤儿有明确行为。
缺点：1）至少一次仍要幂等。2）pika 线程模型易踩。3）RPC 放大队列与通道成本。

适用场景

• 库存预占、风控询问等需要应答的解耦调用。
• 通知类消息的信封规范（即使不用 RPC）。
• 测试要断言配对而非到达顺序。

不适用：用 RPC 替代数据库事务；每请求一条 exclusive 队列打满大促；把 correlation_id 当 message_id 混用。

注意事项

• correlation_id 管对话，message_id 管信本身，两者都要。
• expiration 类型是字符串。
• 4.x 无 immediate。
• 安全：reply_to 不要指向别人的可写队列名（伪造回调）。
• 回调队列权限：库存需能向其 publish，通常默认交换器 + 队列名，VHost 内 write 正则要覆盖生成名（exclusive 名随机，这是正则.*对 reply 的压力）。实践：库存账号对amq.gen-*或改用 direct reply-to 扩展。若正则过死，RPC 回调会 403。订单 VHost 内可给库存amq.gen-.*write，或使用 RabbitMQDirect Reply-to（amq.rabbitmq.reply-to）减少临时队列。测试至少验证一种。

Direct Reply-to 可作加分实验：客户端reply_to=amq.rabbitmq.reply-to，无需 declare exclusive。不展开源码，生产选型时与 exclusive 对比连接亲和。

常见踩坑（生产）

1. 按到达顺序取响应，串单超卖。根因：无 correlation_id。
2. 每次请求 declare 队列，队列数打满。根因：把 exclusive 当请求级。
3. 超时又重试却换了 message_id，库存双扣。根因：幂等键不稳定。

思考题

1. Direct Reply-to 在连接断开时，尚未取走的响应去哪？与 exclusive 队列对比。
2. 若库存 worker 多实例，如何保证同一 message_id 只预占一次？MQ 能代替数据库唯一约束吗？

附录 C：第 11 章思考题参考答案

题 1：跨 VHost 看见同一支付成功。
Shovel/Federation 按权限最小搬运（第 21 章）；或应用双发（要幂等）；不要给订单应用 marketing 的 write。复制的是消息拷贝，不是共享队列。

题 2：topic permission。
可以：普通 write 允许交换机名，topic 权限再限制 routing key 模式。这样能发order.error不能发order.info。测试要单独声明 topic permission，只测普通正则不够。

延伸阅读与资源

SQLAlchemy 2.0从入门到进阶的实战之旅
Dify 从入门到进阶：LLM 应用平台实战修炼
Java 工程师进阶：从 JVM 生产排障到OpenJDK原理
NumPy 从入门到生产落地：全链路实战指南（科学计算/向量化）
Redis 8 实战精讲：从 CRUD 到源码，构建高可用缓存系统
Redis 实战修炼与原理进阶
Python 3实战精进：从脚本到高并发订单引擎
python入门：Rquests从菜鸟脚本到企业级SDK的网络实战圣经
Milvus向量数据库实战修炼：从 0 到 1精通向量检索与生产落地
MongoDB 实战进阶与内核修炼
后端工程师的 AI 转型第一课：Ollama 与私有化大模型实战
10倍开发者的 Dify 魔法书：从零构建全栈 AI 应用
后端工程师转型AI第一课-Ollama 与私有化大模型实战
大型语言模型(LLM) vLLM 高性能推理落地实战
Agent开发之LlamaIndex 实战修炼与源码进阶
大语言模型Transformers 实战修炼与源码剖析