Kafka 消费模式详解：队列 vs 广播

本文详细介绍 Kafka 的两种消费模式，以及与 MQTT 的对比，帮助你在 IoT 项目中选择合适的消息中间件。

📋 目录

• 引言
• Kafka 的两种消费模式
• Consumer Group 详解
• 实战案例
• Kafka vs MQTT
• 最佳实践
• 总结

引言

在使用 Kafka 时，很多人会有这样的疑问：

Kafka 的消息是只能被一个消费者消费，还是像 MQTT 一样以广播形式发给所有消费者？

答案是：Kafka 支持两种模式，可以灵活切换！

这正是 Kafka 相比 MQTT 的一大优势。让我们深入了解这两种模式。

Kafka 的两种消费模式

1. 队列模式（Queue）

特点：每条消息只被一个消费者消费

┌─────────────────────────────────────────┐
│ Topic: device_thing (6 个分区) │
└─────────────────────────────────────────┘
↓
┌─────────────────────────────────────────┐
│ Consumer Group: rule-engine │
│ ├── Consumer 1 → Partition 0, 1, 2 │
│ └── Consumer 2 → Partition 3, 4, 5 │
└─────────────────────────────────────────┘

结果：每条消息只被 Consumer 1 或 Consumer 2 消费一次

使用场景：

• 任务处理（每个任务只需要被处理一次）
• 负载均衡（多个消费者分担负载）
• 顺序处理（同一分区内保证顺序）

代码示例：

@Service
public class OrderProcessService {
@Autowired
private MqConsumer<OrderMessage> consumer;

@PostConstruct
public void init() {
// 所有实例使用同一个 Consumer Group
consumer.consume("order_topic", this::processOrder);
}

private void processOrder(OrderMessage order) {
// 处理订单（每个订单只被处理一次）
System.out.println("处理订单: " + order.getId());
}
}

2. 广播模式（Publish-Subscribe）

特点：每条消息被所有消费者消费

┌─────────────────────────────────────────┐
│ Topic: device_thing │
└─────────────────────────────────────────┘
↓
┌────────────┼────────────┐
↓ ↓ ↓
┌─────────┐ ┌─────────┐ ┌─────────┐
│ Group A │ │ Group B │ │ Group C │
│规则引擎 │ │数据存储 │ │实时监控 │
└─────────┘ └─────────┘ └─────────┘

结果：每条消息被 A、B、C 都消费一次（类似 MQTT）

使用场景：

• 日志收集（多个系统都需要日志）
• 数据同步（同步到多个下游系统）
• 实时监控（多个监控服务）

代码示例：

// 服务 A：规则引擎
@Service
public class RuleEngineService {
@Autowired
private MqConsumer<DeviceMessage> consumer;

@PostConstruct
public void init() {
// Consumer Group: rule-engine
consumer.consume("device_topic", this::processRules);
}
}

// 服务 B：数据存储
@Service
public class DataStorageService {
@Autowired
private MqConsumer<DeviceMessage> consumer;

@PostConstruct
public void init() {
// Consumer Group: data-storage (不同的 Group)
consumer.consume("device_topic", this::saveData);
}
}

// 服务 C：实时监控
@Service
public class MonitoringService {
@Autowired
private MqConsumer<DeviceMessage> consumer;

@PostConstruct
public void init() {
// Consumer Group: monitoring (不同的 Group)
consumer.consume("device_topic", this::monitor);
}
}

效果：

设备消息 → Kafka Topic: device_topic
↓
├── RuleEngineService (Group: rule-engine) ✅ 消费
├── DataStorageService (Group: data-storage) ✅ 消费
└── MonitoringService (Group: monitoring) ✅ 消费

同一条消息被 3 个服务都消费了！

Consumer Group 详解

什么是 Consumer Group？

Consumer Group 是 Kafka 消费模式的核心概念：

• 同一个 Group 内：每条消息只被消费一次（队列模式）
• 不同 Group 之间：每条消息都被消费（广播模式）

Consumer Group 的工作原理

1. 分区分配

Topic: device_thing (6 个分区)
Consumer Group: rule-engine (2 个消费者)

分配策略：
├── Consumer 1 → Partition 0, 1, 2
└── Consumer 2 → Partition 3, 4, 5

如果增加到 3 个消费者：
├── Consumer 1 → Partition 0, 1
├── Consumer 2 → Partition 2, 3
└── Consumer 3 → Partition 4, 5

如果增加到 6 个消费者：
├── Consumer 1 → Partition 0
├── Consumer 2 → Partition 1
├── Consumer 3 → Partition 2
├── Consumer 4 → Partition 3
├── Consumer 5 → Partition 4
└── Consumer 6 → Partition 5

如果增加到 7 个消费者：
├── Consumer 1 → Partition 0
├── Consumer 2 → Partition 1
├── Consumer 3 → Partition 2
├── Consumer 4 → Partition 3
├── Consumer 5 → Partition 4
├── Consumer 6 → Partition 5
└── Consumer 7 → 空闲（无分区可分配）

关键点：

• ✅ 消费者数量 ≤ 分区数量：充分利用
• ⚠️ 消费者数量 > 分区数量：部分消费者空闲

2. 负载均衡

假设每秒 1000 条消息：

1 个消费者：
└── 处理 1000 条/秒（压力大）

2 个消费者：
├── Consumer 1: 500 条/秒
└── Consumer 2: 500 条/秒

6 个消费者：
├── Consumer 1: 167 条/秒
├── Consumer 2: 167 条/秒
├── Consumer 3: 167 条/秒
├── Consumer 4: 167 条/秒
├── Consumer 5: 167 条/秒
└── Consumer 6: 165 条/秒

3. 故障转移

初始状态：
├── Consumer 1 → Partition 0, 1, 2
└── Consumer 2 → Partition 3, 4, 5

Consumer 2 故障：
└── Consumer 1 → Partition 0, 1, 2, 3, 4, 5
（自动接管所有分区）

Consumer 2 恢复：
├── Consumer 1 → Partition 0, 1, 2
└── Consumer 2 → Partition 3, 4, 5
（重新平衡）

实战案例

案例 1：IoT 设备消息处理（队列模式）

场景：100 个网关设备上报数据，需要规则引擎处理

架构：

设备上报
↓
MQTT Broker (EMQX)
↓
Kafka Topic: device_thing (6 个分区)
↓
Consumer Group: rule-engine (3 个实例)
├── Instance 1 → Partition 0, 1
├── Instance 2 → Partition 2, 3
└── Instance 3 → Partition 4, 5
↓
规则引擎处理

代码实现：

@Service
public class RuleDeviceConsumer implements ConsumerHandler<ThingModelMessage> {

@Autowired
private MqConsumer<ThingModelMessage> consumer;

@Autowired
private List<DeviceMessageHandler> handlers;

@PostConstruct
public void init() {
// 所有实例使用同一个 Consumer Group
consumer.consume("device_thing", this);
}

@Override
public void handler(ThingModelMessage msg) {
// 分发给各个 Handler 处理
for (DeviceMessageHandler handler : handlers) {
if (handler.support(msg)) {
handler.handle(msg);
}
}
}
}

优点：

• ✅ 负载均衡：3 个实例分担负载
• ✅ 高可用：任一实例故障，其他实例接管
• ✅ 水平扩展：增加实例即可提升处理能力

案例 2：设备消息多系统处理（广播模式）

场景：设备消息需要同时被规则引擎、数据存储、实时监控处理

架构：

设备上报
↓
Kafka Topic: device_thing
↓
├── Consumer Group: rule-engine → 规则引擎
├── Consumer Group: data-storage → 数据存储
└── Consumer Group: monitoring → 实时监控

代码实现：

// 1. 规则引擎服务
@Service
public class RuleEngineService implements ConsumerHandler<ThingModelMessage> {

@Autowired
private MqConsumer<ThingModelMessage> consumer;

@PostConstruct
public void init() {
consumer.consume("device_thing", this);
}

@Override
public void handler(ThingModelMessage msg) {
// 执行规则引擎逻辑
executeRules(msg);
}
}

// 2. 数据存储服务
@Service
public class DataStorageService implements ConsumerHandler<ThingModelMessage> {

@Autowired
private MqConsumer<ThingModelMessage> consumer;

@Autowired
private DeviceDataRepository repository;

@PostConstruct
public void init() {
consumer.consume("device_thing", this);
}

@Override
public void handler(ThingModelMessage msg) {
// 存储到数据库
repository.save(convertToEntity(msg));
}
}

// 3. 实时监控服务
@Service
public class MonitoringService implements ConsumerHandler<ThingModelMessage> {

@Autowired
private MqConsumer<ThingModelMessage> consumer;

@Autowired
private MetricsCollector metricsCollector;

@PostConstruct
public void init() {
consumer.consume("device_thing", this);
}

@Override
public void handler(ThingModelMessage msg) {
// 收集监控指标
metricsCollector.collect(msg);
}
}

效果：

同一条设备消息：
├── 规则引擎处理 ✅
├── 存储到数据库 ✅
└── 更新监控指标 ✅

优点：

• ✅ 解耦：各服务独立开发和部署
• ✅ 灵活：可以随时添加新的消费服务
• ✅ 可靠：某个服务故障不影响其他服务

案例 3：混合模式（队列 + 广播）

场景：规则引擎需要负载均衡，数据存储需要独立消费

架构：

Kafka Topic: device_thing
↓
├── Consumer Group: rule-engine (3 个实例)
│ ├── Instance 1 → Partition 0, 1
│ ├── Instance 2 → Partition 2, 3
│ └── Instance 3 → Partition 4, 5
│
└── Consumer Group: data-storage (1 个实例)
└── Instance 1 → Partition 0, 1, 2, 3, 4, 5

效果：

设备消息 → Kafka
↓
├── 规则引擎：3 个实例负载均衡处理 ✅
└── 数据存储：1 个实例独立消费 ✅

Kafka vs MQTT

对比表

消费模式对比

MQTT：天然广播

MQTT Topic: device/+/data
↓
├── 订阅者 A ✅ 收到消息
├── 订阅者 B ✅ 收到消息
└── 订阅者 C ✅ 收到消息

特点：
– 所有订阅者都收到消息
– 无法实现队列模式
– 无法负载均衡

Kafka：灵活切换

Kafka Topic: device_data
↓
队列模式（同一个 Group）：
└── Consumer Group: processor
├── Consumer 1 ✅ 处理部分消息
└── Consumer 2 ✅ 处理部分消息

广播模式（不同 Group）：
├── Consumer Group: processor ✅ 消费所有消息
├── Consumer Group: storage ✅ 消费所有消息
└── Consumer Group: monitor ✅ 消费所有消息

特点：
– 两种模式可以共存
– 灵活配置
– 自动负载均衡

何时使用 Kafka？何时使用 MQTT？

使用 Kafka 的场景

✅ 数据流处理

设备 → Kafka → 规则引擎 → 数据库

✅ 日志收集

应用日志 → Kafka → 日志分析系统

✅ 事件驱动架构

订单创建 → Kafka → [支付服务, 库存服务, 通知服务]

✅ 需要消息持久化和回溯

Kafka 保留 7 天数据，可以重新消费

使用 MQTT 的场景

✅ 设备实时通信

设备 ↔ MQTT Broker ↔ 应用

✅ 低延迟要求

传感器数据实时上报（微秒级延迟）

✅ 轻量级协议

资源受限的嵌入式设备

✅ 简单的发布/订阅

不需要复杂的消费模式

混合使用（推荐）

设备
↓ MQTT（低延迟）
MQTT Broker
↓ 桥接
Kafka（持久化、处理）
↓
├── 规则引擎
├── 数据存储
└── 实时监控

优点：

• ✅ MQTT 负责设备通信（低延迟）
• ✅ Kafka 负责数据处理（高吞吐、持久化）
• ✅ 发挥各自优势

最佳实践

1. Consumer Group 命名规范

// ❌ 不推荐：使用随机 ID
String groupId = UUID.randomUUID().toString();

// ✅ 推荐：使用有意义的名称
String groupId = "rule-engine-service";
String groupId = "data-storage-service";
String groupId = "monitoring-service";

// ✅ 推荐：使用类名（自动唯一）
String groupId = handler.getClass().getName().replace(".", "");

2. 分区数量设置

# 根据消费者数量设置分区
消费者数量 = 3 → 分区数量 = 3 或 6
消费者数量 = 5 → 分区数量 = 5 或 10

# 推荐：分区数量 = 消费者数量的倍数
kafka-topics.sh –create \\
–bootstrap-server localhost:9092 \\
–topic device_thing \\
–partitions 6 \\
–replication-factor 3

3. 消费者数量配置

# application.yml
spring:
application:
name: rule–engine–service
kafka:
consumer:
# 每个实例的消费者线程数
concurrency: 3

# 部署 3 个实例，每个实例 3 个线程
# 总共 9 个消费者，建议分区数量 = 9 或 12

4. 监控 Consumer Lag

# 查看消费延迟
kafka-consumer-groups.sh \\
–bootstrap-server localhost:9092 \\
–describe \\
–group rule-engine-service

# 输出示例
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
rule-engine-service device_thing 0 11181 11181 0
rule-engine-service device_thing 1 0 0 0
rule-engine-service device_thing 2 0 0 0
rule-engine-service device_thing 3 0 0 0
rule-engine-service device_thing 4 0 0 0
rule-engine-service device_thing 5 11353 11353 0

# LAG = 0 表示没有积压，消费正常
# LAG > 0 表示有积压，需要增加消费者或优化处理逻辑

5. 故障处理

@Service
public class RobustConsumerService implements ConsumerHandler<Message> {

@Override
public void handler(Message msg) {
try {
// 处理消息
processMessage(msg);
} catch (Exception e) {
// 记录错误日志
log.error("处理消息失败: {}", msg, e);

// 发送到死信队列
sendToDeadLetterQueue(msg, e);

// 或者重试
retryMessage(msg);
}
}

private void sendToDeadLetterQueue(Message msg, Exception e) {
// 发送到专门的错误处理 Topic
producer.send("dead_letter_queue", msg);
}
}

6. 性能优化

// 批量处理
@Service
public class BatchConsumerService {

private List<Message> buffer = new ArrayList<>();
private static final int BATCH_SIZE = 100;

@Override
public void handler(Message msg) {
buffer.add(msg);

if (buffer.size() >= BATCH_SIZE) {
// 批量处理
processBatch(buffer);
buffer.clear();
}
}

@Scheduled(fixedDelay = 1000)
public void flushBuffer() {
if (!buffer.isEmpty()) {
processBatch(buffer);
buffer.clear();
}
}
}

总结

Kafka 消费模式核心要点

选择建议

实战架构

┌─────────────────────────────────────────────────────┐
│ IoT 平台架构 │
├─────────────────────────────────────────────────────┤
│ │
│ 设备层 │
│ └── 设备 (MQTT 客户端) │
│ ↓ MQTT 协议（低延迟） │
│ │
│ 接入层 │
│ └── MQTT Broker (EMQX) │
│ ↓ 桥接到 Kafka │
│ │
│ 消息层 │
│ └── Kafka Cluster │
│ ├── Topic: device_thing (设备上行) │
│ └── Topic: emqx_device_thing (设备下行) │
│ ↓ │
│ │
│ 处理层（广播模式） │
│ ├── Consumer Group: rule-engine │
│ │ └── 规则引擎服务 (3 个实例，队列模式) │
│ │ │
│ ├── Consumer Group: data-storage │
│ │ └── 数据存储服务 (1 个实例) │
│ │ │
│ └── Consumer Group: monitoring │
│ └── 实时监控服务 (1 个实例) │
│ │
│ 存储层 │
│ ├── Redis (实时数据) │
│ ├── MySQL (业务数据) │
│ └── ClickHouse (历史数据) │
│ │
└─────────────────────────────────────────────────────┘

参考资料

• Kafka 官方文档
• Consumer Group 详解
• MQTT 协议规范
• IoT 架构最佳实践

关于作者

本文基于实际 IoT 项目经验总结，涵盖了 Kafka 消费模式的核心概念和最佳实践。

如果你有任何问题或建议，欢迎交流讨论！

版权声明：本文为原创内容，转载请注明出处。

最后更新：2025-12-03