kafka java 多线程_20. 多线程开发者实例
# 多線程 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));
}
}
..
```
總結
以上是生活随笔為你收集整理的kafka java 多线程_20. 多线程开发者实例的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 头发一周洗几次才适宜? - 生活至上，美
- 下一篇： Comparison of relati