最近項目開發過程使用kafka作為項目模塊間負載轉發器，實現實時接收不同產品線消息，分布式準實時消費產品線消息。通過kafka作為模塊間的轉換器，不僅有MQ的幾大好處：異步、?解耦、?削峰等幾大好處，而且開始考慮最大的好處，可以實現架構的水平擴展，下游系統出現性能瓶頸，容器平臺伸縮增加一些實例消費能力很快就提上來了，整體系統架構上不用任何變動。理論上，我們項目數據量再大整體架構上高可用都沒有問題。在使用kafka過程中也遇到一些問題：

1. 消息逐漸積壓，消費能力跟不上；

2.某個消費者實例因為某些異常原因掛掉，造成少量數據丟失的問題。

針對消費積壓的問題，通過研究kafka多線程消費的原理，解決了消費積壓的問題。所以，理解多線程的Consumer模型是非常有必要，對于我們正確處理kafka多線程消費很重要。

kafka多線程消費模式
說kafka多線程消費模式前，我們先來說下kafka本身設計的線程模型和ConcurrentmodificationException異常的原因。見官方文檔：

The Kafka consumer is NOT thread-safe. All network I/O happens in the thread of the application making the call. It is the responsibility of the user to ensure that multi-threaded access is properly synchronized. Un-synchronized access will result in?ConcurrentModificationException.

ConcurrentmodificationException異常的出處見以下代碼：

? /**
? ? ?* Acquire the light lock protecting this consumer from multi-threaded access. Instead of blocking
? ? ?* when the lock is not available, however, we just throw an exception (since multi-threaded usage is not
? ? ?* supported).
? ? ?* @throws IllegalStateException if the consumer has been closed
? ? ?* @throws ConcurrentModificationException if another thread already has the lock
? ? ?*/
? ? private void acquire() {
? ? ? ? ensureNotClosed();
? ? ? ? long threadId = Thread.currentThread().getId();
? ? ? ? if (threadId != currentThread.get() && !currentThread.compareAndSet(NO_CURRENT_THREAD, threadId))
? ? ? ? ? ? throw new ConcurrentModificationException("KafkaConsumer is not safe for multi-threaded access");
? ? ? ? refcount.incrementAndGet();
? ? }
該方法acquire 會在KafkaConsumer的大部分公有方法調用第一句就判斷是否正在同一個KafkaConsumer被多個線程調用。

"正在"怎么理解呢？我們順便看下KafkaConsumer的commitAsync 這個方法就知道了。

?@Override
? ? public void commitAsync(OffsetCommitCallback callback) {
? ? ? ? acquire(); // 引用開始
? ? ? ? try {
? ? ? ? ? ? commitAsync(subscriptions.allConsumed(), callback);
? ? ? ? } finally {
? ? ? ? ? ? release(); //引用釋放
? ? ? ? }
? ? }
我們看KafkaConsumer的release方法就是釋放正在操作KafkaConsumer實例的引用。

?/**
? ? ?* Release the light lock protecting the consumer from multi-threaded access.
? ? ?*/
? ? private void release() {
? ? ? ? if (refcount.decrementAndGet() == 0)
? ? ? ? ? ? currentThread.set(NO_CURRENT_THREAD);
? ? }
通過以上的代碼理解，我們可以總結出來kafka多線程的要點： kafka的KafkaConsumer必須保證只能被一個線程操作。

下面就來說說，我理解的Kafka能支持的兩種多線程模型，首先，我們必須保證操作KafkaConsumer實例的只能是一個線程，那我們要想多線程只能用在消費ConsumerRecord List上動心思了。下面列舉我理解的kafka多線程消費模式。

模式一??1個Consumer模型對應一個線程消費，最多可以有topic對應的partition個線程同時消費Topic。
????????????

?

模式二 1個Consumer和多個線程消費模型，保證只有一個線程操作KafkaConsumer，其它線程消費ConsumerRecord列表。

注意 第二種模式其實也可以支持多個Consumer，用戶最多可以啟用partition總數個Consumer實例，然后，模式二跟模式一唯一的差別就是模式二在單個Consuemr里面是多線程消費，而模式一單個Consumer里面是單線程消費。

以上兩種kafka多線程消費模式優缺點對比：

kafka多線程消費模式實現????
關于多線程消費模式具體實現都是選擇基于spring-kafka實現，畢竟站在巨人肩膀上，站的高望的遠少加班???，以下就是模式二的具體實現，模式一的話就是對模式二的簡化，具體實現如下。

@Configuration
@EnableKafka
public class KafkaConfig {
?
? ? @Value("${kafka.bootstrap-servers}")
? ? private String servers;
?
? ? @Value("${kafka.producer.retries}")
? ? private int retries;
? ? @Value("${kafka.producer.batch-size}")
? ? private int batchSize;
? ? @Value("${kafka.producer.linger}")
? ? private int linger;
?
? ? @Value("${kafka.consumer.enable.auto.commit}")
? ? private boolean enableAutoCommit;
? ? @Value("${kafka.consumer.session.timeout}")
? ? private String sessionTimeout;
? ? @Value("${kafka.consumer.group.id}")
? ? private String groupId;
? ? @Value("${kafka.consumer.auto.offset.reset}")
? ? private String autoOffsetReset;
?
? ? @Value("${msg.consumer.max.poll.records}")
? ? private int maxPollRecords;
?
? ? public Map<String, Object> producerConfigs() {
? ? ? ? Map<String, Object> props = new HashMap<>();
? ? ? ? props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
? ? ? ? props.put(ProducerConfig.RETRIES_CONFIG, retries);
? ? ? ? props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
? ? ? ? props.put(ProducerConfig.LINGER_MS_CONFIG, linger);
? ? ? ? props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
? ? ? ? props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
? ? ? ? return props;
? ? }
?
? ? public ProducerFactory producerFactory() {
? ? ? ? return new DefaultKafkaProducerFactory(producerConfigs());
? ? }
?
? ? @Bean
? ? public KafkaTemplate kafkaTemplate() {
? ? ? ? return new KafkaTemplate(producerFactory());
? ? }
?
? ? @Bean
? ? public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, Object>>
? ? kafkaListenerContainerFactory() {
? ? ? ? ConcurrentKafkaListenerContainerFactory<String, Object> factory =
? ? ? ? ? ? ? ? new ConcurrentKafkaListenerContainerFactory<>();
? ? ? ? factory.setConsumerFactory(consumerFactory());
? ? ? ? factory.setBatchListener(true);
? ? ? ? // 此處并發度設置的都是Consumer個數，可以設置1到partition總數,
? ? ? ? // 但是，所有機器實例上總的并發度之和必須小于等于partition總數
? ? ? ? // 如果，總的并發度小于partition總數，有一個Consumer實例會消費超過一個以上partition
? ? ? ? factory.setConcurrency(2);
? ? ? ? factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
? ? ? ? return factory;
? ? }
?
? ? public ConsumerFactory<String, Object> consumerFactory() {
? ? ? ? return new DefaultKafkaConsumerFactory<>(consumerConfigs());
? ? }
?
? ? public Map<String, Object> consumerConfigs() {
? ? ? ? Map<String, Object> propsMap = new HashMap<>();
? ? ? ? propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
? ? ? ? propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
? ? ? ? propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeout);
? ? ? ? propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
? ? ? ? propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
? ? ? ? propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
? ? ? ? propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
? ? ? ? propsMap.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
? ? ? ? return propsMap;
? ? }
?
}
具體業務代碼在BaseConsumer：

public abstract class BaseConsumer implements ApplicationListener<ConsumerStoppedEvent> {
?
? ? private static final Logger LOG = LoggerFactory.getLogger(BaseConsumer.class);
?
? ? @Value("${kafka.consumer.thread.min}")
? ? private int consumerThreadMin;
?
? ? @Value("${kafka.consumer.thread.max}")
? ? private int consumerThreadMax;
?
? ? private ThreadPoolExecutor consumeExecutor;
?
? ? private volatile boolean isClosePoolExecutor = false;
?
? ? @PostConstruct
? ? public void init() {
?
? ? ? ? this.consumeExecutor = new ThreadPoolExecutor(
? ? ? ? ? ? ? ? getConsumeThreadMin(),
? ? ? ? ? ? ? ? getConsumeThreadMax(),
? ? ? ? ? ? ? ? // 此處最大最小不一樣沒啥大的意義,因為消息隊列需要達到 Integer.MAX_VALUE 才有點作用，
? ? ? ? ? ? ? ? // 矛盾來了，我每次批量拉下來不可能設置Integer.MAX_VALUE這么多，
? ? ? ? ? ? ? ? // 個人覺得每次批量下拉的原則 覺得消費可控就行，
? ? ? ? ? ? ? ? // 不然，如果出現異常情況下，整個服務示例突然掛了，拉下來太多，這些消息會被重復消費一次。
? ? ? ? ? ? ? ? 1000 * 60,
? ? ? ? ? ? ? ? TimeUnit.MILLISECONDS,
? ? ? ? ? ? ? ? new LinkedBlockingQueue<>());
? ? }
?
? ? /**
? ? ?* 收到spring-kafka 關閉Consumer的通知
? ? ?* @param event 關閉Consumer 事件
? ? ?*/
? ? @Override
? ? public void onApplicationEvent(ConsumerStoppedEvent event) {
?
? ? ? ? isClosePoolExecutor = true;
? ? ? ? closeConsumeExecutorService();
?
? ? }
?
? ? private void closeConsumeExecutorService() {
?
? ? ? ? if (!consumeExecutor.isShutdown()) {
?
? ? ? ? ? ? ThreadUtil.shutdownGracefully(consumeExecutor, 120, TimeUnit.SECONDS);
? ? ? ? ? ? LOG.info("consumeExecutor stopped");
?
? ? ? ? }
?
? ? }
?
? ? @PreDestroy
? ? public void doClose() {
? ? ? ? if (!isClosePoolExecutor) {
? ? ? ? ? ? closeConsumeExecutorService();
? ? ? ? }
? ? }
?
? ? @KafkaListener(topics = "${msg.consumer.topic}", containerFactory = "kafkaListenerContainerFactory")
? ? public void onMessage(List<String> msgList, Acknowledgment ack) {
?
? ? ? ? CountDownLatch countDownLatch = new CountDownLatch(msgList.size());
?
? ? ? ? for (String message : msgList) {
? ? ? ? ? ? submitConsumeTask(message, countDownLatch);
? ? ? ? }
?
? ? ? ? try {
? ? ? ? ? ? countDownLatch.await();
? ? ? ? } catch (InterruptedException e) {
? ? ? ? ? ? LOG.error("countDownLatch exception ", e);
? ? ? ? }
?
? ? ? ? // 本次批量消費完,手動提交
? ? ? ? ack.acknowledge();
? ? ? ? LOG.info("finish commit offset");
?
? ? }
?
? ? private void submitConsumeTask(String message, CountDownLatch countDownLatch) {
? ? ? ? consumeExecutor.submit(() -> {
? ? ? ? ? ? try {
? ? ? ? ? ? ? ? onDealMessage(message);
? ? ? ? ? ? } catch (Exception ex) {
? ? ? ? ? ? ? ? LOG.error("on DealMessage exception:", ex);
? ? ? ? ? ? } finally {
? ? ? ? ? ? ? ? countDownLatch.countDown();
? ? ? ? ? ? }
? ? ? ? });
? ? }
?
? ? /**
? ? ?* 子類實現該抽象方法處理具體消息的業務邏輯
? ? ?* @param message kafka的消息
? ? ?*/
? ? protected abstract void onDealMessage(String message);
?
? ? private int getConsumeThreadMax() {
? ? ? ? return consumerThreadMax;
? ? }
?
? ? private int getConsumeThreadMin() {
? ? ? ? return consumerThreadMin;
? ? }
?
? ? public void setConsumerThreadMax(int consumerThreadMax) {
? ? ? ? this.consumerThreadMax = consumerThreadMax;
? ? }
?
? ? public void setConsumerThreadMin(int consumerThreadMin) {
? ? ? ? this.consumerThreadMin = consumerThreadMin;
? ? }
}
其中，closeConsumeExecutorService方法就是為了服務實例異常退出或者多機房上線kill的情況下，盡最大可能保證本次拉下來的任務被消費掉。最后，附上closeConsumeExecutorService實現，覺得RocketMQ源碼這個實現的不錯，就借用過來了，在此表示感謝。

? public static void shutdownGracefully(ExecutorService executor, long timeout, TimeUnit timeUnit) {
? ? ? ? // Disable new tasks from being submitted.
? ? ? ? executor.shutdown();
? ? ? ? try {
? ? ? ? ? ? // Wait a while for existing tasks to terminate.
? ? ? ? ? ? if (!executor.awaitTermination(timeout, timeUnit)) {
? ? ? ? ? ? ? ? executor.shutdownNow();
? ? ? ? ? ? ? ? // Wait a while for tasks to respond to being cancelled.
? ? ? ? ? ? ? ? if (!executor.awaitTermination(timeout, timeUnit)) {
? ? ? ? ? ? ? ? ? ? LOGGER.warn(String.format("%s didn't terminate!", executor));
? ? ? ? ? ? ? ? }
? ? ? ? ? ? }
? ? ? ? } catch (InterruptedException ie) {
? ? ? ? ? ? // (Re-)Cancel if current thread also interrupted.
? ? ? ? ? ? executor.shutdownNow();
? ? ? ? ? ? // Preserve interrupt status.
? ? ? ? ? ? Thread.currentThread().interrupt();
? ? ? ? }
? ? }
下面回到使用kafka遇到的第二個問題，怎么解決消費者實例因為某些原因掛掉，造成少量數據丟失的問題。其實，通過我們上面的寫法，已經不會出現因為某些原因服務實例（docker、物理機）掛掉，丟數據的情況。因為我們是先拉取后消費，消費完才手動提交kafka確認offset。實在還存在萬一退出時候調用的closeConsumeExecutorService方法還沒有消費完數據，表示這個時候offset肯定沒有手動提交，這一部分數據也不會丟失，會在服務實例恢復了重新拉取消費。

以上的代碼存在極小的可能瑕疵，比如，我們雙機房切換上線，某機房實例有一部分數據沒有消費，下次會重復消費的問題。其實，這個問題我們在業務上通過在配置中心配置一個標識符來控制，當改變標識符控制某些機房停止拉取kafka消息，這個時候我們就可以安全操作，不擔心kafka沒有消費完，下次重復消費的問題了。
————————————————
版權聲明：本文為CSDN博主「Johnniecsdn」的原創文章，遵循CC 4.0 BY-SA版權協議，轉載請附上原文出處鏈接及本聲明。
原文鏈接：https://blog.csdn.net/Johnnyz1234/article/details/98318528

總結

以上是生活随笔為你收集整理的正确处理kafka多线程消费的姿势的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。