# 多线程 Consumer Instance

## Kafka Java Consumer 设计原理

* Kafka Java Consumer 是单线程设计

* 从 Kafka V0.10.1.0，KafkaConsumer 是双线程：用户主线程 & 心跳线程

* 用户主线程

* 启动 Consumer 应用 main 方法的线程

* 心跳线程

* 只负责定期给对应的 Broker 发送心跳，标示 Consumer 的存活性(liveness)

* 新版本设计：单线程 + 轮询机制：

* 实现非阻塞式的消息获取

## 多线程方案

* KafkaConsumer 类不是 thread-safe

* 所有的网络 IO 处理都是发生在用户主线程中

* 不能在多个线程中共享同一个 KafkaConsumer 实例

* 可以使用 `KafkaConsumer.wakeup()` 在其他线程中唤醒 Consumer

基于非 thread-safe，两套多线程方案

* 消费者程序启动多个线程，每个线程维护专属的 KafkaConsumer Instance，负责完整的消息获取、消息处理流程

* 消费者程序使用单或多线程获取消息，同时创建多个消费线程执行消息处理逻辑

* 处理消息交由特定的线程池来做

* 将消息获取与处理解耦


## Code

### 方案 1

```

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 consumer;

private ExecutorService executors;

...

private int workerNum = ...;

executors = new ThreadPoolExecutor(

workerNum, workerNum, 0L, TimeUnit.MILLISECONDS,

new ArrayBlockingQueue<>(1000),

new ThreadPoolExecutor.CallerRunsPolicy());

...

while (true) {

ConsumerRecords records =

consumer.poll(Duration.ofSeconds(1));

for (final ConsumerRecord record : records) {

executors.submit(new Worker(record));

}

}

..

```