流控、过滤与 ACL

一句话：这四件事各管一层——Tag/SQL92 决定「哪些消息可见」，三层流控决定「写不写得进来」，消息轨迹决定「消息去过哪」，ACL 决定「谁有权限」。

适用版本：4.9 / 5.3 / 5.5。SQL92 过滤需要 Broker 显式开启；ACL 配置文件名与权限模型在 4.x/5.x 一致。

为什么需要

默认配置下，Broker 接受任何客户端的连接，任何 Consumer 都能订阅任何 Topic，出问题只能靠翻应用日志。生产环境需要四个能力：让 Consumer 只收到自己关心的消息；在 Broker 过载时快速失败；能回答「这条消息谁发的、谁消费的」；能限制谁能发、谁能收。

这四件事经常被混在一起讲，但它们的执行位置和代价完全不同，配置时不能互相替代。

核心机制

一、过滤：Tag 与 SQL92

方式	执行位置	依据	代价
Tag	Broker	ConsumeQueue 中的 8 字节 Tag hash	低，先按 hash 过滤，再读消息
SQL92	Broker	消息属性（properties）	中，需要解析表达式并逐条判断
消费端过滤	Consumer	消息体或属性	高，无关消息也占带宽和拉取配额

// 一个 Topic 一个订阅表达式；多个 Tag 用 || 组合
consumer.subscribe("order-topic", "PAID || SHIPPED");
consumer.subscribe("order-topic", "*");                 // 全部

// 一个 Consumer 可以订阅多个 Topic：多次调用 subscribe
consumer.subscribe("payment-topic", "SUCCESS");

// SQL92：按属性过滤，Broker 需 enablePropertyFilter=true
consumer.subscribe("order-topic",
    MessageSelector.bySql("price > 100 AND color = 'red'"));

关键约束：

• 同一个 Consumer 对同一个 Topic 只能有一个订阅表达式。对同一 Topic 再调一次 subscribe，后一次会覆盖前一次，不会取并集。
• 想匹配多个 Tag，用一个表达式加 ||，不要拆成多次订阅。
• 一个 Consumer 可以订阅多个 Topic，每个 Topic 各有自己的表达式。
• SQL92 依赖消息属性，发送时必须 putUserProperty；属性不会自动从消息体解析。

Tag 过滤为什么便宜：ConsumeQueue 每条 20 字节里带 Tag 的 hash，Broker 可以先按 hash 跳过不匹配的条目，不必读 CommitLog。hash 冲突时还会读消息做精确比对，所以不会误投。

二、三层流控

层	触发条件	表现	相关参数
PageCache	写入 PageCache 的等待超过阈值	Broker 返回 SYSTEM_BUSY	osPageCacheBusyTimeOutMills（默认 1000ms）
Broker 线程池	发送请求排队满	快速失败，Producer 收到错误	sendMessageThreadPoolNums（4.x 默认 1，5.x 以版本为准）
Producer	发送超时或 Broker 不可用	同步发送重试，异步回调异常	retryTimesWhenSendFailed（默认 2）、sendMsgTimeout（默认 3000ms）

PageCache 繁忙通常意味着磁盘写入跟不上：磁盘 IO 打满、刷盘线程阻塞，或消息体过大。此时 Broker 拒绝新写入是保护自己，不是 bug。

Producer 收到 SYSTEM_BUSY 或超时后会自动重试；重试仍失败就进入业务兜底。不要在业务层无脑无限重试，否则会把 Broker 的恢复窗口也打满。

消费侧还有一层背压：pullThresholdForQueue（默认 1000 条）和 pullThresholdSizeForQueue（默认 100 MiB）限制单 Queue 本地缓存，超过就暂停拉取。这不是 Broker 流控，但能防止 Consumer 被压垮。

三、消息轨迹

开启后，一条消息会产生三类记录：Producer 的发送信息、Broker 的存储位置、Consumer 的消费耗时与结果。

# broker.conf
traceTopicEnable = true
# 可选：自定义轨迹 Topic，默认 RMQ_SYS_TRACE_TOPIC
# traceTopicName = RMQ_SYS_TRACE_TOPIC

producer.setEnableMsgTrace(true);
consumer.setEnableMsgTrace(true);
// 可选：producer.setCustomizedTraceTopic("RMQ_SYS_TRACE_TOPIC");

msgId 是查轨迹的入口；轨迹数据本身也写在系统 Topic 里，会占用磁盘和写入带宽。开销比例取决于消息大小、发送频率和保留时间，没有通用的「5%～10%」数字，必须在自己环境压测。排查期临时开启、定位后关闭是常见做法。

四、ACL

ACL 在 Broker 端拦截，不是客户端 SDK 层。即使客户端改代码，Broker 也会拒绝未授权的请求。

# broker.conf
aclEnable = true
# 默认读取 ${ROCKETMQ_HOME}/conf/plain_acl.yml
# aclConfigPath = /opt/rocketmq/conf/plain_acl.yml

最小配置：

globalWhiteRemoteAddresses: [10.0.0.*]     # 内网网段白名单，按实际环境改
accounts:
  - {accessKey: order-producer, secretKey: change-me-1, defaultTopicPerm: DENY, topicPerms: [order-topic=PUB]}
  - {accessKey: order-consumer, secretKey: change-me-2, defaultTopicPerm: DENY, topicPerms: [order-topic=SUB]}
  - {accessKey: ops-admin, secretKey: change-me-3, admin: true}

权限值：DENY、PUB、SUB、PUB|SUB、ANY。admin: true 的账号拥有管理权限，只给运维工具使用。

AclClientRPCHook hook = new AclClientRPCHook(
    new SessionCredentials("order-producer", "change-me-1"));
DefaultMQProducer producer = new DefaultMQProducer("pg", hook);

Consumer 同理，在构造时传入 AclClientRPCHook。修改 plain_acl.yml 后 Broker 会检测文件变化并重载（以版本为准）；不确定时滚动重启最稳妥。

最小上手

# 1. 打开 ACL、轨迹、SQL92
cat >> conf/broker-local.conf <<'EOF'
aclEnable = true
traceTopicEnable = true
enablePropertyFilter = true
EOF

# 2. 写入最小 ACL 文件（沿用上面的格式，换成测试账号）
cat > conf/plain_acl.yml <<'EOF'
globalWhiteRemoteAddresses: [127.0.0.1]
accounts:
  - {accessKey: demo, secretKey: demo-secret, defaultTopicPerm: DENY, topicPerms: [demo=PUB|SUB]}
EOF

# 3. 重启 Broker；不带凭证的客户端应被拒绝
sh bin/mqadmin topicStatus -n 127.0.0.1:9876 -t demo

# 4. 发送三条 Tag 为 PAID / SHIPPED / OTHER 的消息，
#    Consumer 订阅 "PAID || SHIPPED"，应只收到两条

SQL92 验证：给消息设置属性 price=200、color=red，订阅 price > 100 AND color = 'red'，再发一条 price=50 的消息，确认收不到。

生产动作

• Tag 设计成有限枚举，不要用用户 ID、订单号做 Tag；Tag 数量爆炸会让订阅表达式和排查都变复杂。
• SQL92 只在必要时开，并给属性建固定 schema；表达式里避免函数和复杂运算，减少 Broker CPU。
• 不要指望「一个 Consumer 订阅同一 Topic 的多个表达式」。合并成一个 || 表达式，或拆成多个 ConsumerGroup。
• ConsumerGroup 不能改名。换组名等于换消费身份，新组按 consumeFromWhere（默认从最新）开始，旧组位点不会带过去；切换前按时间重置位点并接受重复。
• ACL 先白名单、后账号：先用 globalWhiteRemoteAddresses 限制网段，再给应用分配最小权限；admin 账号不落到业务应用。轨迹按需开启，流控参数按下游能力调。

边界

• Kafka 是 Consumer Pull，不是「Broker 把所有消息推给 Consumer」。Kafka Consumer 主动 fetch，过滤通常在客户端做；RocketMQ 的 Tag/SQL92 过滤在 Broker 端。
• Tag 过滤不是内容过滤。它只看 Tag，不看消息体；按消息体字段过滤要用 SQL92 属性或消费端处理。
• ACL 不是传输加密。ACL 管「谁能访问」，TLS 管「传输过程是否加密」；金融场景两者都要。
• 轨迹有成本，会增加 Broker 写入和磁盘占用，具体开销随流量变化，不能拍固定百分比；流控触发也不代表消息丢，Producer 会重试，重试耗尽仍需业务落库补偿。
• 换 ConsumerGroup 会丢位点。这不是 ACL 或过滤的问题，但经常和「订阅不生效」一起出现，排查时先确认组名没变。

讲清楚

过滤、流控、轨迹、ACL 是四件独立的事。过滤决定 Consumer 能看到哪些消息：Tag 在 Broker 端用 ConsumeQueue 里的 hash 先筛，便宜；SQL92 按消息属性筛，灵活但费 CPU；一个 Consumer 对同一个 Topic 只能有一个表达式，多 Tag 用 ||，多个 Topic 可以分别订阅。

流控分三层：PageCache 忙、Broker 发送线程池满、Producer 重试，分别在 Broker 和客户端拒绝或重试。轨迹记录生产、存储、消费三段，用 msgId 查，有额外写入和存储成本，开销要实测。ACL 是 Broker 端权限，用 AccessKey/SecretKey 加 Topic 级 PUB/SUB 权限，客户端改代码也绕不过去。几个容易说错的点：Kafka 是 Consumer 主动 Pull，不是 Broker 推送；ACL 不负责加密；ConsumerGroup 改名会丢位点。把这些边界分清，配置才不会互相打架。