[TOC] #### 1. 前言 --- 前面四讲我们学了四种 Exchange，这一讲我们把它们串起来，做一个完整的实战：日志分级系统 先思考一下，不同级别日志为什么要不同处理 ？ 一个线上系统，每天会产生大量不同类型或级别的日志： + 用户登录、浏览商品 → info 普通日志 + 接口响应慢、内存偏高 → warning 警告日志 + 数据库连接失败、支付异常 → error 错误日志 这三种日志，处理方式完全不同： | 级别 | 处理方式 | | ------------ | ------------ | | info | 存起来就行，用于后续分析 | | warning | 存起来 + 统计监控，关注趋势 | | error | 存起来 + 立刻告警，发短信通知运维 | 如果所有日志都走同一条路，要么告警太多（info 也报警），要么告警太少（error 被淹没） 你需要的是：不同级别的日志，走不同的处理流程 #### 2. 架构设计 --- 我们用 Topic Exchange 来实现： ```plaintext Producer 发日志 → log_exchange (Topic) → error_queue → 告警系统 → warning_queue → 监控系统 → info_queue → 日志系统 → all_queue → 全量归档 ``` Producer 发消息时，用 `模块.级别` 作为路由标识： + order.error + user.warning + pay.info 各队列用通配符绑定： + \*.error → error_queue + \*.warning → warning_queue + \*.info → info_queue + \# → all_queue #### 3. 代码实现 --- 声明队列和绑定 ```php // 声明 Topic 类型的 Exchange $channel->exchange_declare('log_exchange', 'topic', false, true, false); // 声明四个队列 $channel->queue_declare('error_queue', false, true, false, false); $channel->queue_declare('warning_queue', false, true, false, false); $channel->queue_declare('info_queue', false, true, false, false); $channel->queue_declare('all_queue', false, true, false, false); // 用通配符绑定 $channel->queue_bind('error_queue', 'log_exchange', '*.error'); $channel->queue_bind('warning_queue', 'log_exchange', '*.warning'); $channel->queue_bind('info_queue', 'log_exchange', '*.info'); $channel->queue_bind('all_queue', 'log_exchange', '#'); ``` Producer：发送日志 ```php // 发一条 error 日志 $message = new AMQPMessage(json_encode([ 'module' => 'order', 'level' => 'error', 'message' => '数据库连接超时', 'time' => date('Y-m-d H:i:s'), ]), [ 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT, 'content_type' => 'application/json', ]); $channel->basic_publish($message, 'log_exchange', 'order.error'); ``` Consumer：各取所需 ```php // 告警系统：只消费 error_queue $callback = function ($msg) { $data = json_decode($msg->body, true); // 发送告警短信 sendAlert("【告警】{$data['module']}模块：{$data['message']}"); $msg->ack(); }; $channel->basic_consume('error_queue', '', false, false, false, false, $callback); ``` 验证一下，分别发送三条不同级别的日志： ```php $channel->basic_publish(new AMQPMessage('订单创建成功'), 'log_exchange', 'order.info'); $channel->basic_publish(new AMQPMessage('接口响应慢'), 'log_exchange', 'user.warning'); $channel->basic_publish(new AMQPMessage('支付失败'), 'log_exchange', 'pay.error'); ``` 然后看管理界面： + info_queue：1 条（订单创建成功） + warning_queue：1 条（接口响应慢） + error_queue：1 条（支付失败） + all_queue：3 条（全部） 各队列各收各的，互不影响 #### 4. 本文小结 --- 这个案例用到了前面学的知识： + Topic Exchange：用通配符按级别分流 + 消息持久化：日志不能丢 + 手动 ACK：处理完再确认 实际项目中，日志系统还会配合 ELK（Elasticsearch + Logstash + Kibana）做存储和可视化 这里只是用日志系统举例帮助理解 Topic Exchange，真实项目中的日志方案有很多种，不一定必须使用 RabbitMQ RabbitMQ 在其中的角色就是 “日志中转站”，收集日志，分发给不同的处理系统 下一讲，我们来聊一个很实际的问题：订单 30 分钟不付款怎么办 ？