@KafkaListener怎么使用

這篇文章主要介紹“@KafkaListener怎么使用”，在日常操作中，相信很多人在@KafkaListener怎么使用問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”@KafkaListener怎么使用”的疑惑有所幫助！接下來，請跟著小編一起來學習吧！

@KafkaListener 如何使用

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

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

在注解內(nèi)指定topic名稱，當對應(yīng)的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注解的類，之后這些類的相關(guān)信息會被注冊到 KafkaListenerEndpointRegistry內(nèi)，包括注解所在的方法，當前的bean等。KafkaListenerEndpointRegistry這個類內(nèi)部會維護多個Listener Container，每一個@KafkaListener都會對應(yīng)一個Listener Container。并且每個Container對應(yīng)一個線程。

監(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的新特性之一，注解可以重復(fù)標注。

@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）假設(shè)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è)消費者數(shù)量為N，當前主題剩下的分區(qū)數(shù)量為M，則當前消費者應(yīng)該分配的分區(qū)數(shù)量 = M%N==0? M/N +1 : M/N ；按照公式，3個消費者應(yīng)該分配的分區(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；

3. Sticky

kafka在0.11版本引入了Sticky分區(qū)分配策略，它的兩個主要目的是：

1. 分區(qū)的分配要盡可能的均勻，分配給消費者者的主題分區(qū)數(shù)最多相差一個；

2. 分區(qū)的分配盡可能的與上次分配的保持相同；

當兩者發(fā)生沖突時，第一個目標優(yōu)先于第二個目標；

粘性分區(qū)是由Kafka從0.11x版本開始引入的分配策略，首先會盡量均衡的分配分區(qū)到消費者上面，在出現(xiàn)同一消費組內(nèi)消費者出現(xiàn)問題的時候，會盡量保持原來的分配的分區(qū)不變；

Sticky分區(qū)初始分配分區(qū)的方法與Range相似，但是不同；拿7個分區(qū)3個消費者為例，消費者消費的分區(qū)依舊是3/2/2，但是不同與Range的是Range分區(qū)是排好序的，但是Sticky分區(qū)是隨機的。

到此，關(guān)于“@KafkaListener怎么使用”的學習就結(jié)束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習，快去試試吧！若想繼續(xù)學習更多相關(guān)知識，請繼續(xù)關(guān)注億速云網(wǎng)站，小編會繼續(xù)努力為大家?guī)砀鄬嵱玫奈恼拢?/p>