Kafka 多线程消费者
- 多线程方案
Kafka 0.10.1.0 后，Kafka Consumer 变为双线程的设计 :
- 用户主线程 : 启动 Consumer 的 main
- 心跳线程 (Heartbeat Thread) : 定期对 Broker 发送心跳请求，探测消费者的存活性 (liveness）
- 将心跳频率与主线程处理频率分开，对俩者进行解耦
老 Consumer 是多线程的架构 :
- Fetcher 线程 : 每个 Consumer 给所有订阅的主题分区创建对应的消息获取线程
- Consumer 是阻塞式的（blocking），Consumer 启动后，内部会创建很多阻塞式的消息获取迭代器
新 Consumer 设计了单线程 + 轮询的机制 , 实现非阻塞式的消息获取
多线程方案
KafkaConsumer 不是线程安全的 (thread-safe)
- 所有的网络 I/O 处理都在用户主线程中，使用时要确保线程安全
- 不能在多个线程中共享同个 Kafka Consumer，不然会 ConcurrentModificationException 异常
两套多线程方案 :
- 消费者启动多个线程，每个线程维护个 KafkaConsumer ，负责完整的消息获取、消息处理流程
- 消费者用单或多线程获取消息，同时创建多个消费线程执行消息处理逻辑
例子 : 消费者要做 1、2、3、4、5
- 方法 1 : 工作划分为粗粒度，会创建多个线程，每个线程要执行 1、2、3、4、5，来实现并行处理，不能分割具体的子任务
- 方法 2 : 更细粒度化，将 1、2 分割，用单线程（也可多线程），将 3、4、5，用别的多个线程
| 方案 | 优点 | 缺点 |
|---|---|---|
| 多线程 + 多 KafkaConsumer | 方便实现 | 占更多系统资源 |
| 速度块, 无线程间交互开销 | 线程数受限于主题分区数, 扩展性差 | |
| 易于维护分区内的消费顺序 | 线程处理消息易超时 , 会 Rebalance | |
| 单线程 + 单 KafkaConsumer + 消息处理 Worker 线程池 | 可独立扩展消费获取线程数和 Worker 线程数 | 实现难度高 |
| 可扩展性好 | 难维护分区的消息消费顺序 | |
| 处理链路长, 不易于位移提交管理 |
方案 1 优势 :
- 实现简单，用多个线程在每个线程中创建各自 KafkaConsumer
- 多个线程之间没有任何交互，能避开线程安全
- 每个线程用各自 KafkaConsumer 来执行消息获取和消息处理逻辑，所以主题中的每个分区都能保证只被一个线程处理，容易实现分区内的消息消费顺序
方案 1 缺点 :
- 每个线程都要维护自己的 KafkaConsumer ，会占用更多的系统资源，如 : 内存、TCP 连接
- 线程数受限于 Consumer 订阅主题的总分区数 : 一个消费者组中，每个分区只能被组内的一个消费者所消费
- 当一个消费者组订阅 100 个分区，那最多只能扩展到 100 个线程
- 每个线程要执行消息获取和消息处理逻辑。一旦消息处理逻辑重，就造成消息处理速度慢，导致 Rebalance，从而引发整个消费者组的消费停滞
方案 2 优势 :
- 将任务分为消息获取和消息处理的俩个线程
- 高伸缩性 : 能独立调节消息获取的线程数，消息处理的线程数，不用考虑两者是否相互影响
- 消费获取速度慢，就增加消费获取的线程数
- 消息的处理速度慢，就增加 Worker 线程池线程数
方案 2 缺点 :
- 实现难度大，要管理两组线程
- 将消息获取和消息处理分开，无法保证分区内的消费顺序。如 : 在某个分区中，消息 1 在消息 2 前被保存，但Worker 线程可能先处理消息 2，再处理消息 1，会破坏消息在分区的顺序
- 多组线程，让整个消息消费链路被拉长，位移提交困难，可能出现重复消费
方案 1 的代码 :
- 创建 Runnable 类 : 执行消费获取和消费处理的逻辑
- 每个 KafkaConsumerRunner 类都会创建各种 KafkaConsumer
public class KafkaConsumerRunner implements Runnable {
private final AtomicBoolean closed = new AtomicBoolean(false);
private final KafkaConsumer consumer;
public void run() {
try {
consumer.subscribe(Arrays.asList("topic"));
while (!closed.get()) {
ConsumerRecords records =
consumer.poll(Duration.ofMillis(10000));
// 执行消息处理逻辑
}
} catch (WakeupException e) {
// Ignore exception if closing
if (!closed.get()) throw e;
} finally {
consumer.close();
}
}
// Shutdown hook which can be called from a separate thread
public void shutdown() {
closed.set(true);
consumer.wakeup();
}
}
方案 2 :
- 由线程池负责处理具体的消息
private final KafkaConsumer<String, String> consumer;
private ExecutorService executors;
//...
private int workerNum = ...;
executors = new ThreadPoolExecutor(
workerNum, workerNum, 0L, TimeUnit.MILLISECONDS,
new ArrayBlockingQueue<>(1000),
new ThreadPoolExecutor.CallerRunsPolicy());
//...
while (true) {
ConsumerRecords<String, String> records =
consumer.poll(Duration.ofSeconds(1));
for (final ConsumerRecord record : records) {
executors.submit(new Worker(record));
}
}
//..
