@KafkaListener 如何使用

更新時間：2023年02月20日 09:23:32 作者：DemonHunter211

這篇文章主要介紹了@KafkaListener 如何使用，本文通過圖文實例代碼相結合給大家詳細講解，文末給大家介紹了kafka的消費者分區(qū)分配策略，需要的朋友可以參考下

@KafkaListener 如何使用

spring-kafka使用基于@KafkaListener注解，@KafkaListener使用方式如下

@KafkaListener(topics = "topic1")
public void? ?kafkaListen(List<ConsumerRecord<xxx, xxx>> records) {
? ? ...
}

在注解內(nèi)指定topic名稱，當對應的topic內(nèi)有新的消息時，testListen方法會被調(diào)用，參數(shù)就是topic內(nèi)新的消息。這個過程是異步進行的。

@KafkaListener工作流程主要有以下幾步:

解析；解析@KafkaListener注解。
注冊；解析后的數(shù)據(jù)注冊到spring-kafka。
監(jiān)聽；開始監(jiān)聽topic變更。
調(diào)用；調(diào)用注解標識的方法，將監(jiān)聽到的數(shù)據(jù)作為參數(shù)傳入。
下面我們一步一步分析

解析

@KafkaListener注解由KafkaListenerAnnotationBeanPostProcessor類解析，后者實現(xiàn)了BeanPostProcessor接口，這個接口如下

public interface BeanPostProcessor {

    Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException;

    Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException;
}

接口內(nèi)部有2個方法，分別在bean初始化前后被調(diào)用。

KafkaListenerAnnotationBeanPostProcessor內(nèi)會在postProcessAfterInitialization方法內(nèi)解析@KafkaListener注解。

注冊
解析步驟里，我們可以獲取到所有含有@KafkaListener注解的類，之后這些類的相關信息會被注冊到 KafkaListenerEndpointRegistry內(nèi)，包括注解所在的方法，當前的bean等。KafkaListenerEndpointRegistry這個類內(nèi)部會維護多個Listener Container，每一個@KafkaListener都會對應一個Listener Container。并且每個Container對應一個線程。

監(jiān)聽
注冊完成之后，每個Listener Container會開始工作，會新啟一個新的線程，初始化KafkaConsumer，監(jiān)聽topic變更等。

調(diào)用
監(jiān)聽到數(shù)據(jù)之后，container會組織消息的格式，隨后調(diào)用解析得到的@KafkaListener注解標識的方法，將組織后的消息作為參數(shù)傳入方法，執(zhí)行用戶邏輯。

@KafkaListener和@KafkaListners

@KafkaListeners是@KafkaListener的Container Annotation，這也是jdk8的新特性之一，注解可以重復標注。

@KafkaListeners({@KafkaListener(topics="topic1"), @KafkaListener(topics="topic2")})
public void listen(ConsumerRecord<Integer, String> msg) {}
 
等同于
 
@KafkaListener(topics="topic1")
@KafkaListener(topics="topic2")
public void listen(ConsumerRecord<Integer, String> msg) {}

擴展：kafka的消費者分區(qū)分配策略

kafka有三種分區(qū)分配策略

1. RoundRobin

2. Range

3. Sticky

1. RoundRobin

（1）把所有topic的分區(qū)partition放入一個隊列中，按照name的hashcode進行排序；

（2）把consumer放在一個循環(huán)隊列，按照name的hashcode進行排序；

（3）循環(huán)遍歷consumer，從partition隊列pop出一個partition，分配給當前consumer；以此類推，取下一個consumer，繼續(xù)從partition隊列pop出來分配給當前consumer；直到partition隊列中的元素被分配完；

2. Range

（1）假設topicA有4個分區(qū)，topicB有5個分區(qū)，topicC有6個分區(qū)；一共有3個consumer；

（2）遍歷3個topic的分區(qū)集合，先取topicA的分區(qū)集合，然后準備依次給3個consumer分配分區(qū)；對于第1個consumer，所分配的分區(qū)數(shù)量根據(jù)以下公式：假設消費者數(shù)量為N，當前主題剩下的分區(qū)數(shù)量為M，則當前消費者應該分配的分區(qū)數(shù)量 = M%N==0? M/N +1 : M/N ；按照公式，3個消費者應該分配的分區(qū)數(shù)量依次為：2/1/1，即topicA-partition-0/1分配給consumer-0，topicA-partition-2分配給consumer-1，topicA-partition-3分配給consumer-2；

（3）按照上述規(guī)則按序把topicB和topicC的分區(qū)分配給3個consumer；依次為：2/2/1，2/2/2；