主题Topic与队列机制：售货柜多设备消息隔离设计
作者：黒漂技术佬
系列专栏：RocketMQ核心原理与无人售货柜项目实战
一、Topic：消息的一级分类
1.1 Topic是什么
Topic（主题）是RocketMQ中最顶层的消息分类单位。你可以把它类比为数据库里的表——消息是表里的行，每条消息都属于某张表。
数据库类比： 数据库 → RocketMQ 数据库 → Broker 表 → Topic 行 → Message 分区 → Queue 列标签 → Tag一个Broker上可以存放多个Topic，每个Topic存放一类业务消息。比如售货柜项目里：
| Topic名 | 用途 | 生产者 | 消费者 |
|---|---|---|---|
| order_topic | 订单消息 | 订单服务 | 库存服务、推送服务 |
| payment_topic | 支付消息 | 支付服务 | 柜子网关、ERP同步服务 |
| device_topic | 设备消息 | 柜子网关 | 监控服务、告警服务 |
| log_topic | 日志消息 | 各服务 | 日志收集服务 |
1.2 Topic的创建方式
自动创建：Producer第一次向一个不存在的Topic发消息时，Broker会自动创建它。开发环境方便，但生产环境强烈建议关闭（autoCreateTopicEnable=false），原因有三：
- 自动创建的Topic默认4个队列，可能不符合业务需求
- 容易因拼写错误创建出错误的Topic（
order_topicvsorder_topc），消息发到错误地方排查困难 - 自动创建的Topic均匀分布在所有Broker上，不可控
手动创建：通过Dashboard或命令行预先创建，可以指定队列数、所在Broker等：
# 命令行创建Topicshmqadmin updateTopic\-n127.0.0.1:9876\-b127.0.0.1:10911\-torder_topic\-r8\# 读队列数-w8# 写队列数也可以用Dashboard界面操作：主题 → 新增 → 填写Topic名和队列数。
1.3 读写队列的含义
创建Topic时会指定读队列数（r）和写队列数（w），这俩有什么区别？
- 写队列（WriteQueue）：Producer发消息时，Broker按写队列数做路由分配，消息实际存在这些队列里
- 读队列（ReadQueue）：Consumer消费时，按读队列数做负载均衡，从这些队列拉消息
正常情况下读队列数 = 写队列数。什么时候会不一样？Topic缩容。假设原来8个队列想缩到4个，直接改写队列数为4，读队列数暂时保持8，等Consumer把原8个队列的消息消费完，再把读队列数改成4。这样缩容不会丢消息。
二、Queue：Topic下的子分区
2.1 Queue的作用
Queue（队列）是Topic的子分区，类似数据库表的分区。一个Topic默认有4个队列（可配置）。
Topic: order_topic (4个队列) Queue-0 ──→ [msg1] [msg5] [msg9] ... Queue-1 ──→ [msg2] [msg6] [msg10] ... Queue-2 ──→ [msg3] [msg7] [msg11] ... Queue-3 ──→ [msg4] [msg8] [msg12] ...Producer发消息时，默认轮询（Round Robin）把消息均匀分配到各队列。Queue的两个核心作用：
并行消费：多个Consumer可以分别消费不同Queue，实现并行处理。1个Topic有8个Queue，最多8个Consumer同时消费，吞吐量线性扩展。
负载均衡：ConsumerGroup内的Consumer实例自动分配Queue，谁消费哪个Queue由Rebalance算法决定。某个Consumer挂了，它的Queue会被重新分配给其他Consumer。
2.2 队列数怎么定
队列数不是越多越好，也不是越少越好。经验法则：
| 场景 | 建议队列数 | 原因 |
|---|---|---|
| 低频消息（订单） | 4~8 | 消费者实例少，多了也用不上 |
| 中频消息（设备状态） | 8~16 | 多个区域消费者并行 |
| 高频消息（日志/埋点） | 16~32 | 高并发需要更多并行度 |
| 顺序消息 | 按业务分区键数量定 | 保证同一Key的消息在同一Queue |
售货柜项目建议：订单Topic 8个队列（8个消费实例够用），设备消息Topic 16个队列（按区域分配），日志Topic 32个队列（高吞吐）。
三、Tag：消息的二级分类
3.1 Tag的概念
Tag（标签）是Topic下的二级分类，用于在同一个Topic内区分子类消息。如果Topic是数据库的表，那Tag就是表里的一个分类字段。
Topic: device_topic ├── Tag: heartbeat （设备心跳消息） ├── Tag: alert （设备告警消息） ├── Tag: status （设备状态消息） └── Tag: inventory （设备库存消息）为什么不用多个Topic代替Tag？因为Topic是物理隔离，每个Topic占独立的存储和队列资源。用Tag在同一Topic下分类，共享队列资源，减少Topic数量，降低管理成本。
3.2 Tag的使用
Producer端指定Tag：
// Topic:Tag 格式rocketMQTemplate.syncSend("device_topic:heartbeat",heartbeatMsg);rocketMQTemplate.syncSend("device_topic:alert",alertMsg);rocketMQTemplate.syncSend("device_topic:status",statusMsg);Consumer端按Tag过滤消费：
// 只消费告警消息@RocketMQMessageListener(topic="device_topic",selectorExpression="alert",// 只消费Tag=alert的消息consumerGroup="alert_consumer_group")publicclassAlertConsumerimplementsRocketMQListener<AlertMessage>{@OverridepublicvoidonMessage(AlertMessagemessage){alertService.handle(message);}}// 消费心跳和状态消息（多Tag用 || 分隔）@RocketMQMessageListener(topic="device_topic",selectorExpression="heartbeat || status",consumerGroup="monitor_consumer_group")publicclassMonitorConsumerimplementsRocketMQListener<MessageExt>{@OverridepublicvoidonMessage(MessageExtmessage){Stringtag=message.getTags();if("heartbeat".equals(tag)){handleHeartbeat(message);}elseif("status".equals(tag)){handleStatus(message);}}}// 消费所有Tag@RocketMQMessageListener(topic="device_topic",selectorExpression="*",// *表示消费所有TagconsumerGroup="all_device_consumer_group")publicclassAllDeviceConsumerimplementsRocketMQListener<MessageExt>{// ...}3.3 Tag vs Topic的选择标准
什么时候用不同Topic，什么时候用不同Tag？记住一个原则：
- 消费方不同、需要物理隔离→ 用不同Topic
- 消费方相同或部分相同、逻辑分类→ 用同一Topic + 不同Tag
举例：
| 场景 | 选择 | 原因 |
|---|---|---|
| 订单消息 vs 支付消息 | 不同Topic | 消费方完全不同，物理隔离 |
| 设备心跳 vs 设备告警 | 同Topic不同Tag | 都属于设备消息，监控服务都要消费 |
| 支付成功 vs 支付失败 | 同Topic不同Tag | 都是支付消息，下游消费逻辑接近 |
四、ConsumerGroup与队列分配关系
4.1 队列分配规则
在集群消费模式下，一个ConsumerGroup内的多个Consumer实例分摊Topic的所有Queue。核心规则：一个Queue同一时间只被组内一个Consumer实例消费。
Topic: order_topic (4个Queue) ConsumerGroup: order_consumer_group 情况1：2个Consumer实例 Consumer-1 ← Queue-0, Queue-1 Consumer-2 ← Queue-2, Queue-3 情况2：4个Consumer实例 Consumer-1 ← Queue-0 Consumer-2 ← Queue-1 Consumer-3 ← Queue-2 Consumer-4 ← Queue-3 情况3：6个Consumer实例（超过队列数！） Consumer-1 ← Queue-0 Consumer-2 ← Queue-1 Consumer-3 ← Queue-2 Consumer-4 ← Queue-3 Consumer-5 ← 空闲（分不到队列） Consumer-6 ← 空闲（分不到队列）4.2 消费者超过队列数怎么办
如上所示，当Consumer实例数 > Queue数时，多出来的Consumer空闲，不消费任何消息。这不是Bug，是设计如此——Queue是并行消费的最小单位，4个Queue最多4个Consumer并行。
所以部署消费服务时，实例数不要超过Topic的Queue数，否则浪费资源。如果需要更多并行度，先增加Queue数。
4.3 Rebalance机制
ConsumerGroup内的Consumer实例数变化时（扩容/缩容/宕机），RocketMQ会自动触发Rebalance（重平衡），重新分配Queue。
初始状态： Consumer-1 ← Queue-0, Queue-1 Consumer-2 ← Queue-2, Queue-3 Consumer-2宕机 → 触发Rebalance： Consumer-1 ← Queue-0, Queue-1, Queue-2, Queue-3 （全部接管） 新Consumer-3加入 → 触发Rebalance： Consumer-1 ← Queue-0, Queue-1 Consumer-3 ← Queue-2, Queue-3Rebalance由Consumer端发起，每20秒检查一次。如果发现队列分配发生变化，自动调整。这个过程对用户透明，但有一个注意点：Rebalance瞬间可能出现消息重复投递（Consumer切换队列时，上一次未确认的消息会被重新投递），所以消费端一定要做幂等。
五、售货柜多设备消息隔离实战方案
5.1 问题背景
假设我们有以下业务需求：
- 全国有10000台售货柜，分布在500个门店
- 每台柜子定时上报心跳、库存、状态
- 柜子关门后上报订单消息
- 柜子异常时上报告警消息
- 不同门店的消息需要隔离处理（A店的运维只关心A店的设备）
- 柜子出货消息要保证同一台设备的顺序性
5.2 隔离方案设计
方案一：按门店ID区分Topic
Topic: store_10001_device_topic (门店10001的设备消息) Topic: store_10002_device_topic (门店10002的设备消息) ...- 优点：物理隔离彻底，不同门店互不影响
- 缺点：500个门店 = 500个Topic，Topic数量爆炸，管理成本高，RocketMQ建议单Broker Topic数不超过5000但太多影响性能
方案二：按设备ID分配队列 + 消息Key
这是推荐的方案。用统一的Topic，通过Queue分配和消息Key来实现逻辑隔离：
Topic: device_message (16个Queue) ├── 用Tag区分消息类型：heartbeat / alert / status / inventory ├── 用设备ID作为消息Key，便于查询 └── 用MessageQueueSelector把同一设备的消息路由到同一QueueProducer端路由：
@ServicepublicclassDeviceMessageService{@ResourceprivateRocketMQTemplaterocketMQTemplate;/** * 发送设备消息（同一设备的消息路由到同一队列，保证顺序） */publicvoidsendDeviceMessage(StringdeviceId,Stringtag,Objectpayload){DeviceMessagemessage=newDeviceMessage(deviceId,tag,payload);// 使用hashKey路由：同一deviceId的消息始终进入同一QueuerocketMQTemplate.syncSendOrderly("device_message:"+tag,// Topic:TagMessageBuilder.withPayload(message).build(),deviceId// hashKey：按设备ID做hash选队列);}}syncSendOrderly方法内部用MessageQueueSelector，对 deviceId 取hash后对队列数取模，保证同一设备的消息始终进同一队列。这样同一设备的消息被同一Consumer消费，保证了消息顺序性。
Consumer端按门店过滤：
@Component@RocketMQMessageListener(topic="device_message",selectorExpression="alert || status",// 只消费告警和状态consumerGroup="store_monitor_group",consumeMode=ConsumeMode.CONCURRENTLY)publicclassStoreMonitorConsumerimplementsRocketMQListener<DeviceMessage>{@OverridepublicvoidonMessage(DeviceMessagemessage){StringdeviceId=message.getDeviceId();// 从设备ID查出所属门店StringstoreId=deviceService.getStoreId(deviceId);// 按门店分发处理StoreHandlerhandler=storeHandlerMap.get(storeId);if(handler!=null){handler.handle(message);}}}方案三：按消息类型用Tag区分 + 按区域用ConsumerGroup
Topic: device_message Tag: heartbeat → ConsumerGroup: heartbeat_group (全国心跳汇总) Tag: alert → ConsumerGroup: alert_group_north (北方区域告警) ConsumerGroup: alert_group_south (南方区域告警) Tag: status → ConsumerGroup: status_group (状态监控) Tag: inventory → ConsumerGroup: inventory_group (库存同步)不同ConsumerGroup各自消费全量消息，在Consumer内部按区域/门店过滤处理。这种方式灵活但ConsumerGroup多，注意不要超过RocketMQ的订阅组限制（默认1000个）。
5.3 最终推荐方案
综合考虑，售货柜项目的消息隔离方案如下：
┌─────────────────────────────────────────────────────────┐ │ Topic 设计 │ ├─────────────────────────────────────────────────────────┤ │ │ │ order_topic (8队列) │ │ └─ Tag: order_created / order_paid / order_closed │ │ │ │ device_message (16队列) │ │ └─ Tag: heartbeat / alert / status / inventory │ │ └─ hashKey: deviceId (保证同设备消息顺序) │ │ │ │ payment_callback (8队列) │ │ └─ Tag: wechat / alipay │ │ │ │ device_log (32队列) │ │ └─ Tag: operation / error / access │ │ └─ 单向发送，不走顺序 │ │ │ ├─────────────────────────────────────────────────────────┤ │ ConsumerGroup 设计 │ ├─────────────────────────────────────────────────────────┤ │ │ │ order_topic: │ │ inventory_consumer_group (库存服务，集群模式) │ │ push_consumer_group (推送服务，集群模式) │ │ │ │ device_message: │ │ alert_consumer_group (告警服务) │ │ monitor_consumer_group (监控服务，消费heartbeat+status) │ │ inventory_sync_group (库存同步服务，消费inventory) │ │ │ │ payment_callback: │ │ gateway_consumer_group (柜子网关，消费后通知出货) │ │ erp_sync_consumer_group (ERP同步服务) │ │ │ └─────────────────────────────────────────────────────────┘5.4 关键设计决策总结
| 设计决策 | 选择 | 理由 |
|---|---|---|
| 门店隔离方式 | 消息Key + Consumer内过滤 | 避免Topic爆炸，逻辑隔离够用 |
| 设备消息顺序 | hashKey=deviceId路由到同一Queue | 出货和库存变动需保序 |
| 消息类型区分 | Tag | 同类设备消息共享Topic，减少Topic数 |
| 消费并行度 | Queue数 = 预计最大Consumer实例数 | 避免实例空闲浪费 |
| 幂等保障 | 订单ID/设备ID+时间戳做去重 | 防止Rebalance导致重复消费 |
| 日志类消息 | 独立Topic + 单向发送 | 和业务消息隔离，互不影响 |
六、小结
这一篇从Topic、Queue、Tag三个维度拆解了RocketMQ的消息分类和分区机制，重点讲解了Queue的并行消费和负载均衡作用、Tag的二级分类过滤、ConsumerGroup与Queue的分配关系。最后给出了一套完整的售货柜多设备消息隔离方案：按业务域分Topic、按消息类型分Tag、按设备ID做Queue路由保证顺序、按消费方分ConsumerGroup。这套方案在后面的系列文章中会持续用到。
