## 前言

消息队列是微服务间异步通信的核心。中台化后，多个业务线共用一套消息队列，如何隔离、如何管理、如何保证可靠性是关键。本篇讲解 Kafka 和 RocketMQ 的中台化建设。

---

## 一、消息队列在中台中的角色

```
中台架构中的消息队列：

  用户中心 → 发布事件 → ┌──────────┐ → 订阅 → 消息中心
  订单中心 → 发布事件 → │  MQ 中台 │ → 订阅 → 库存中心
  支付中心 → 发布事件 → │          │ → 订阅 → 推荐引擎
  商品中心 → 发布事件 → └──────────┘ → 订阅 → 数据中台

  职责：
  1. 事件驱动 → 中心间解耦
  2. 削峰填谷 → 应对流量洪峰
  3. 最终一致 → 跨中心数据同步
  4. 日志采集 → 统一日志管道
```

---

## 二、Kafka vs RocketMQ 选型

| 维度 | Kafka | RocketMQ |
|------|-------|----------|
| 吞吐量 | 极高（百万级） | 高（十万级） |
| 延迟 | 毫秒级 | 毫秒级 |
| 顺序消息 | 分区内有序 | 分区内有序 |
| 事务消息 | 弱 | 强 |
| 延迟消息 | 不支持 | 支持（18 级） |
| 死信队列 | 不支持 | 支持 |
| 消息回溯 | 支持 | 支持 |
| 多租户 | Topic 隔离 | Group 隔离 |
| 运维 | 较复杂 | 较简单 |
| 生态 | 流处理 | 消息业务 |

### 选型建议

```yaml
# 选 Kafka 如果你：
  - 高吞吐日志/数据管道
  - 需要流处理（Kafka Streams/Flink）
  - 已有 Kafka 基础设施

# 选 RocketMQ 如果你：
  - 业务消息（订单/支付）
  - 需要事务消息
  - 需要延迟消息
  - 需要死信队列
```

---

## 三、Kafka 中台化

### 集群架构

```
┌─────────────────────────────────────────────────┐
│              Kafka Cluster (3 Broker)            │
│                                                  │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐      │
│  │ Broker 0 │  │ Broker 1 │  │ Broker 2 │      │
│  │          │  │          │  │          │      │
│  │ Topic A  │  │ Topic A  │  │ Topic B  │      │
│  │ (Leader) │  │(Follower)│  │ (Leader) │      │
│  │ Topic C  │  │ Topic B  │  │ Topic C  │      │
│  │          │  │          │  │          │      │
│  └──────────┘  └──────────┘  └──────────┘      │
│                                                  │
│  Zookeeper (3 节点)                              │
│  Schema Registry                                 │
│  Kafka Connect                                   │
│  Kafka Streams                                   │
└─────────────────────────────────────────────────┘
```

### 多租户隔离

```yaml
# 方案 1：Topic 前缀隔离（推荐）
Topic 命名规范:
  格式: <team>.<domain>.<event>
  
  示例:
    payments.team.payment.completed
    payments.team.payment.failed
    orders.team.order.created
    inventory.team.stock.updated

# 方案 2：独立集群隔离（高安全级别）
  团队 A → Kafka Cluster A
  团队 B → Kafka Cluster B
```

### Topic 管理

```bash
# 创建 Topic
kafka-topics.sh --create \
  --bootstrap-server kafka:9092 \
  --topic payments.payment.completed \
  --partitions 6 \
  --replication-factor 3 \
  --config retention.ms=604800000 \
  --config max.message.bytes=1048576

# 查看所有 Topic
kafka-topics.sh --list \
  --bootstrap-server kafka:9092

# 查看 Topic 详情
kafka-topics.sh --describe \
  --bootstrap-server kafka:9092 \
  --topic payments.payment.completed

# 生产消息
kafka-console-producer.sh \
  --bootstrap-server kafka:9092 \
  --topic payments.payment.completed

# 消费消息
kafka-console-consumer.sh \
  --bootstrap-server kafka:9092 \
  --topic payments.payment.completed \
  --from-beginning \
  --group payments-consumer
```

### 配额限流

```bash
# 按 Client ID 限流
kafka-configs.sh --bootstrap-server kafka:9092 \
  --alter --add-config \
  'producer_byte_rate=10485760,consumer_byte_rate=10485760' \
  --entity-type clients \
  --entity-name payments-client

# 按 User 限流（需要 SASL）
kafka-configs.sh --bootstrap-server kafka:9092 \
  --alter --add-config \
  'producer_byte_rate=5242880,consumer_byte_rate=5242880' \
  --entity-type users \
  --entity-name orders-user
```

### ACL 权限控制

```bash
# 授权 payments-team 只能读写 payments.* Topic
kafka-acls.sh --bootstrap-server kafka:9092 \
  --add --allow-principal User:payments-team \
  --producer --topic "payments." \
  --resource-prefix

kafka-acls.sh --bootstrap-server kafka:9092 \
  --add --allow-principal User:payments-team \
  --consumer --topic "payments." \
  --resource-prefix \
  --group "payments-"
```

### Schema Registry

```bash
# 注册 Avro Schema
curl -X POST http://schema-registry:8081/subjects/payments.completed-value/versions \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{
    "schema": "{\"type\":\"record\",\"name\":\"PaymentCompleted\",\"fields\":[{\"name\":\"paymentId\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"}]}"
  }'

# 生产端使用 Schema
# 消费端自动校验 Schema
```

### Kafka Connect 数据管道

```json
// 从 MySQL 同步到 Kafka
{
  "name": "mysql-orders-source",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "mysql",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "dbz",
    "database.server.id": "184054",
    "database.allowPublicKeyRetrieval": "true",
    "database.server.name": "orders",
    "table.include.list": "orders.orders,orders.order_items",
    "database.history.kafka.bootstrap.servers": "kafka:9092",
    "database.history.kafka.topic": "schema-changes.orders"
  }
}
```

---

## 四、RocketMQ 中台化

### 部署

```bash
# NameServer
docker run -d --name rmqnamesrv \
  -p 9876:9876 \
  apache/rocketmq:5.2.0 \
  sh mqnamesrv

# Broker
docker run -d --name rmqbroker \
  -p 10911:10911 -p 10909:10909 \
  --link rmqnamesrv:namesrv \
  -e "NAMESRV_ADDR=namesrv:9876" \
  apache/rocketmq:5.2.0 \
  sh mqbroker -c /home/rocketmq/rocketmq-5.2.0/conf/broker.conf
```

### Topic 管理

```bash
# 创建 Topic
mqadmin updateTopic -n namesrv:9876 \
  -b broker-a \
  -t payments.payment.completed \
  -r 3 -w 3    # 3 读 3 写队列

# 创建消费组
mqadmin updateSubGroup -n namesrv:9876 \
  -b broker-a \
  -g payments-consumer-group \
  -r 2    # 重试 2 次
```

### 事务消息

```java
// 事务消息生产者
TransactionMQProducer producer = new TransactionMQProducer("tx-group");
producer.setNamesrvAddr("namesrv:9876");

// 事务监听器
producer.setTransactionListener(new TransactionListener() {
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        // 执行本地事务
        try {
            orderService.createOrder(msg);
            return LocalTransactionState.COMMIT_MESSAGE;
        } catch (Exception e) {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }
    
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        // 事务回查
        Order order = orderService.findById(msg.getKeys());
        if (order != null && order.isCreated()) {
            return LocalTransactionState.COMMIT_MESSAGE;
        }
        return LocalTransactionState.UNKNOW;
    }
});

// 发送事务消息
Message msg = new Message("orders.created", 
    "order-data".getBytes());
producer.sendMessageInTransaction(msg, null);
```

### 延迟消息

```java
// 延迟消息（RocketMQ 支持 18 个延迟级别）
Message msg = new Message("orders.timeout", 
    "order-data".getBytes());
// 延迟级别: 1=1s 2=5s 3=10s ... 18=2h
msg.setDelayTimeLevel(4);  // 30s 后投递
producer.send(msg);
```

### 消费者组管理

```yaml
# 消费者配置
consumer:
  group_name: payments-consumer-group
  namesrv_addr: namesrv:9876
  consume_mode: CONSUME_PASSIVELY  # 被动消费
  consume_from_where: CONSUME_FROM_LAST_OFFSET  # 从最后位置消费
  max_reconsume_times: 3   # 重试 3 次
  suspend_current_queue_time_millis: 1000  # 暂停时间
```

### 死信队列处理

```bash
# 死信队列 Topic 格式：%DLQ%<consumer_group>
# 查看
mqadmin consumerProgress -n namesrv:9876 -g payments-consumer-group

# 消费死信队列
mqadmin queryMsgByOffset -n namesrv:9876 \
  -b broker-a -t %DLQ%payments-consumer-group \
  -o <offset>
```

---

## 五、消息可靠性保证

### 生产端可靠性

```python
# Kafka：acks=all
from kafka import KafkaProducer
producer = KafkaProducer(
    bootstrap_servers=['kafka:9092'],
    acks='all',              # 所有副本确认
    retries=3,               # 失败重试
    max_in_flight_requests_per_connection=1,  # 顺序保证
    enable_idempotence=True, # 幂等生产
    compression_type='lz4',  # 压缩
)

# 发送并等待确认
future = producer.send('payments.completed', value=message)
result = future.get(timeout=10)  # 同步等待
```

```java
// RocketMQ：同步发送
DefaultMQProducer producer = new DefaultMQProducer("group");
producer.setNamesrvAddr("namesrv:9876");
producer.setRetryTimesWhenSendFailed(3);    // 重试 3 次
producer.setSendMsgTimeout(5000);           // 超时 5 秒
producer.start();

SendResult result = producer.send(msg);
if (result.getSendStatus() != SendStatus.SEND_OK) {
    // 处理发送失败
}
```

### 消费端可靠性

```python
# Kafka 消费者
from kafka import KafkaConsumer
consumer = KafkaConsumer(
    'payments.completed',
    bootstrap_servers=['kafka:9092'],
    group_id='payments-consumer',
    enable_auto_commit=False,    # 手动提交
    auto_offset_reset='earliest',
)

for message in consumer:
    try:
        process_message(message)
        consumer.commit()    # 处理成功后提交
    except Exception as e:
        # 处理失败，不提交，下次重新消费
        log.error(f"Failed: {e}")
```

### 消息幂等

```python
# 消费幂等性设计
def process_payment_message(message):
    payment_id = message['payment_id']
    
    # 幂等检查
    if is_processed(payment_id):
        return  # 已处理，跳过
    
    # 处理业务
    process_payment(message)
    
    # 标记已处理
    mark_as_processed(payment_id)
```

---

## 六、监控告警

### Kafka 监控

```yaml
# Kafka Exporter
scrape_configs:
  - job_name: 'kafka'
    static_configs:
      - targets: ['kafka-exporter:9308']

# 关键指标
# kafka_consumergroup_lag → 消费延迟
# kafka_topic_partitions → 分区数
# kafka_topic_partition_under_replicated_partition → 欠同步分区
```

### 告警规则

```yaml
groups:
  - name: mq-alerts
    rules:
      - alert: KafkaConsumerLag
        expr: kafka_consumergroup_lag > 10000
        for: 10m
        labels:
          severity: warning
      
      - alert: KafkaUnderReplicated
        expr: kafka_topic_partition_under_replicated_partition > 0
        for: 5m
        labels:
          severity: critical
      
      - alert: RocketMQBrokerDown
        expr: rocketmq_broker_runtime_status == 0
        for: 1m
        labels:
          severity: critical
```

---

## 要点回顾

| 维度 | Kafka | RocketMQ |
|------|-------|----------|
| 适用 | 高吞吐/流处理 | 业务消息 |
| 多租户 | Topic 前缀 | Group 隔离 |
| 事务消息 | 弱 | 强 |
| 延迟消息 | 不支持 | 支持 |
| 死信队列 | 不支持 | 支持 |
| 配额限流 | Client ID | Consumer |

- Kafka 适合数据管道，RocketMQ 适合业务消息
- 多租户用 Topic 前缀命名规范
- 生产端 acks=all + 重试 + 幂等
- 消费端手动提交 + 幂等处理
- 监控重点：Consumer Lag 和分区同步状态

---

## 下一篇预告

下一篇 **【中台·技术篇】缓存中台建设：Redis 集群统一管理与多级缓存架构** 将讲解缓存中台的建设。

消息队列是微服务间异步通信的核心。中台化后，多个业务线共用一套消息队列，如何隔离、如何管理、如何保证可靠性是关键。本篇讲解 Kafka 和 RocketMQ 的中台化建设。

【中台·技术篇】消息队列中台：Kafka/RocketMQ 统一管理与多租户隔离

研发部经理

后端

架构设计

云计算

消息队列中台化建设核心解析：对比Kafka与RocketMQ在微服务异步通信中的选型要点，详述Topic前缀隔离、ACL权限控制、事务消息与死信队列实现，以及生产消费端可靠性保障、幂等处理和监控告警配置，助力企业构建高可用MQ中台架构。

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

【中台·技术篇】消息队列中台：Kafka/RocketMQ 统一管理与多租户隔离-腾讯云开发者社区-腾讯云

【中台·技术篇】消息队列中台：Kafka/RocketMQ 统一管理与多租户隔离

【中台·技术篇】消息队列中台：Kafka/RocketMQ 统一管理与多租户隔离

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐