spring-Kafka中的@KafkaListener深入源碼解讀
更新時間：2023年02月20日 09:45:33 作者：柏油
本文主要通過深入了解源碼，梳理從spring啟動到真正監(jiān)聽kafka消息的這套流程，從spring啟動開始處理@KafkaListener，本文結(jié)合實例流程圖給大家講解的非常詳細，需要的朋友參考下
前言
本文主要通過深入了解源碼，梳理從spring啟動到真正監(jiān)聽kafka消息的這套流程
一、總體流程
從spring啟動開始處理@KafkaListener，到start消息監(jiān)聽整體流程圖
二、源碼解讀
1、postProcessAfterInitialization
KafkaListenerAnnotationBeanPostProcessor#postProcessAfterInitialization
public Object postProcessAfterInitialization(final Object bean, final String beanName) throws BeansException {
if (!this.nonAnnotatedClasses.contains(bean.getClass())) {
Class<?> targetClass = AopUtils.getTargetClass(bean);
// 掃描@KafkaListener注解
Collection<KafkaListener> classLevelListeners = findListenerAnnotations(targetClass);
......
if (annotatedMethods.isEmpty()) {
this.nonAnnotatedClasses.add(bean.getClass());
this.logger.trace(() -> "No @KafkaListener annotations found on bean type: " + bean.getClass());
}
else {
// Non-empty set of methods
for (Map.Entry<Method, Set<KafkaListener>> entry : annotatedMethods.entrySet()) {
Method method = entry.getKey();
// 遍歷掃描到的所有@KafkaListener注解并開始處理
for (KafkaListener listener : entry.getValue()) {
processKafkaListener(listener, method, bean, beanName);
}
}
this.logger.debug(() -> annotatedMethods.size() + " @KafkaListener methods processed on bean '"
+ beanName + "': " + annotatedMethods);
}
// 處理在類上的@KafkaListener注解
if (hasClassLevelListeners) {
processMultiMethodListeners(classLevelListeners, multiMethods, bean, beanName);
}
}
return bean;
}
1.1、processKafkaListener
KafkaListenerAnnotationBeanPostProcessor#processKafkaListener
protected void processKafkaListener(KafkaListener kafkaListener, Method method, Object bean, String beanName) {
Method methodToUse = checkProxy(method, bean);
MethodKafkaListenerEndpoint<K, V> endpoint = new MethodKafkaListenerEndpoint<>();
endpoint.setMethod(methodToUse);
processListener(endpoint, kafkaListener, bean, methodToUse, beanName);
}1.2、processListener
KafkaListenerAnnotationBeanPostProcessor#processListener
將每個kafkaListener轉(zhuǎn)變成MethodKafkaListenerEndpoint并注冊到KafkaListenerEndpointRegistrar容器，方便后續(xù)統(tǒng)一啟動監(jiān)聽
protected void processListener(MethodKafkaListenerEndpoint<?, ?> endpoint, KafkaListener kafkaListener,
Object bean, Object adminTarget, String beanName) {
String beanRef = kafkaListener.beanRef();
if (StringUtils.hasText(beanRef)) {
this.listenerScope.addListener(beanRef, bean);
}
endpoint.setBean(bean);
endpoint.setMessageHandlerMethodFactory(this.messageHandlerMethodFactory);
endpoint.setId(getEndpointId(kafkaListener));
endpoint.setGroupId(getEndpointGroupId(kafkaListener, endpoint.getId()));
endpoint.setTopicPartitions(resolveTopicPartitions(kafkaListener));
endpoint.setTopics(resolveTopics(kafkaListener));
endpoint.setTopicPattern(resolvePattern(kafkaListener));
endpoint.setClientIdPrefix(resolveExpressionAsString(kafkaListener.clientIdPrefix(), "clientIdPrefix"));
String group = kafkaListener.containerGroup();
......
// 注冊已經(jīng)封裝好的消費端-endpoint
this.registrar.registerEndpoint(endpoint, factory);
if (StringUtils.hasText(beanRef)) {
this.listenerScope.removeListener(beanRef);
}
}1.3、registerEndpoint
KafkaListenerEndpointRegistrar#registerEndpoint
public void registerEndpoint(KafkaListenerEndpoint endpoint, KafkaListenerContainerFactory<?> factory) {
......
KafkaListenerEndpointDescriptor descriptor = new KafkaListenerEndpointDescriptor(endpoint, factory);
synchronized (this.endpointDescriptors) {
// 如果到了需要立即啟動監(jiān)聽的階段就直接注冊并監(jiān)聽(也就是創(chuàng)建消息監(jiān)聽容器并啟動)
if (this.startImmediately) { // Register and start immediately
this.endpointRegistry.registerListenerContainer(descriptor.endpoint,
resolveContainerFactory(descriptor), true);
}
else {
// 一般情況都先走這一步，添加至此列表，待bean后續(xù)的生命周期 統(tǒng)一注冊并啟動
this.endpointDescriptors.add(descriptor);
}
}
}
public void registerListenerContainer(KafkaListenerEndpoint endpoint, KafkaListenerContainerFactory<?> factory,
boolean startImmediately) {
......
synchronized (this.listenerContainers) {
......
// 1.創(chuàng)建消息監(jiān)聽容器
MessageListenerContainer container = createListenerContainer(endpoint, factory);
this.listenerContainers.put(id, container);
if (StringUtils.hasText(endpoint.getGroup()) && this.applicationContext != null) {
List<MessageListenerContainer> containerGroup;
if (this.applicationContext.containsBean(endpoint.getGroup())) {
containerGroup = this.applicationContext.getBean(endpoint.getGroup(), List.class);
}
else {
containerGroup = new ArrayList<MessageListenerContainer>();
this.applicationContext.getBeanFactory().registerSingleton(endpoint.getGroup(), containerGroup);
}
containerGroup.add(container);
}
// 2.是否立即啟動消息監(jiān)聽
if (startImmediately) {
startIfNecessary(container);
}
}
}1.4、startIfNecessary
KafkaListenerEndpointRegistry#startIfNecessary
啟動消息監(jiān)聽
private void startIfNecessary(MessageListenerContainer listenerContainer) {
if (this.contextRefreshed || listenerContainer.isAutoStartup()) {
// 啟動消息監(jiān)聽
// 到這一步之后，消息監(jiān)聽以及處理都是KafkaMessageListenerContainer的邏輯
// 到此也就打通了@KafkaListener到MessageListenerContainer消息監(jiān)聽容器的邏輯
listenerContainer.start();
}
}2、afterSingletonsInstantiated
這一步是實例化(此處的實例化是已經(jīng)創(chuàng)建對象并完成了初始化操作)之后，緊接著的操作
KafkaListenerAnnotationBeanPostProcessor#afterSingletonsInstantiated
public void afterSingletonsInstantiated() {
this.registrar.setBeanFactory(this.beanFactory);
// 對"注冊員"信息的完善
if (this.beanFactory instanceof ListableBeanFactory) {
Map<String, KafkaListenerConfigurer> instances =
((ListableBeanFactory) this.beanFactory).getBeansOfType(KafkaListenerConfigurer.class);
for (KafkaListenerConfigurer configurer : instances.values()) {
configurer.configureKafkaListeners(this.registrar);
}
}
if (this.registrar.getEndpointRegistry() == null) {
if (this.endpointRegistry == null) {
Assert.state(this.beanFactory != null,
"BeanFactory must be set to find endpoint registry by bean name");
this.endpointRegistry = this.beanFactory.getBean(
KafkaListenerConfigUtils.KAFKA_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME,
KafkaListenerEndpointRegistry.class);
}
this.registrar.setEndpointRegistry(this.endpointRegistry);
}
......
// Actually register all listeners
// 整個方法這里才是關(guān)鍵
// 創(chuàng)建MessageListenerContainer并注冊
this.registrar.afterPropertiesSet();
}
2.1、afterPropertiesSet
KafkaListenerEndpointRegistrar#afterPropertiesSet
public void afterPropertiesSet() {
registerAllEndpoints();
}2.2、registerAllEndpoints
KafkaListenerEndpointRegistrar#registerAllEndpoints
protected void registerAllEndpoints() {
synchronized (this.endpointDescriptors) {
for (KafkaListenerEndpointDescriptor descriptor : this.endpointDescriptors) {
// 這里是真正的創(chuàng)建ListenerContainer監(jiān)聽對象并注冊
this.endpointRegistry.registerListenerContainer(
descriptor.endpoint, resolveContainerFactory(descriptor));
}
// 啟動時所有消息監(jiān)聽對象都注冊之后，便將參數(shù)置為true
this.startImmediately = true; // trigger immediate startup
}
}總結(jié)
以上便是整個流程，總體感覺就是將kafka消息監(jiān)聽融入到spring生命周期中，并完美契合
調(diào)試及相關(guān)源碼版本：
org.springframework.boot:：2.3.3.RELEASE spring-kafka:2.5.4.RELEASE
相關(guān)參考：
到此這篇關(guān)于spring-Kafka中的@KafkaListener深入源碼解讀的文章就介紹到這了,更多相關(guān)spring-Kafka @KafkaListener內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
SpringBoot整合RabbitMQ實現(xiàn)延遲隊列的示例詳解
這篇文章主要為大家詳細介紹了SpringBoot如何整合RabbitMQ實現(xiàn)延遲隊列，文中的示例代碼講解詳細，具有一定的學習價值，感興趣的可以了解一下2023-04-04
SpringSecurity6.4中一次性令牌登錄(One-Time Token Login)實現(xiàn)
Spring Security為一次性令牌認證提供了支持,本文就來介紹一下SpringSecurity6.4中一次性令牌登錄(One-Time Token Login)實現(xiàn),具有一定的參考價值,感興趣的可以了解一下2025-03-03
今天帶大家學習Java并發(fā)編程的相關(guān)知識,文中對Fork/Join框架作了非常詳細的介紹,對正在學習有關(guān)知識的小伙伴們很有幫助,需要的朋友可以參考下2021-06-06
這篇文章主要為大家詳細介紹了Java實現(xiàn)斗地主案例，文中示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2020-04-04
這篇文章主要介紹了使用@NonNull注解遇到的小問題及解決，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2023-01-01
這篇文章主要介紹了java如何遍歷對象中的所有屬性(字段)和類型問題，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2023-07-07
java數(shù)據(jù)結(jié)構(gòu)與算法之簡單選擇排序詳解
這篇文章主要介紹了java數(shù)據(jù)結(jié)構(gòu)與算法之簡單選擇排序,結(jié)合實例形式分析了選擇排序的原理、實現(xiàn)方法與相關(guān)操作技巧,需要的朋友可以參考下2017-05-05
這篇文章主要為大家詳細介紹了Java實現(xiàn)簡易圖書借閱系統(tǒng)，文中示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2022-03-03
這篇文章主要介紹了idea安裝hsdis,本文給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2021-03-03
這篇文章主要介紹了MyBatis按時間排序方式，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2022-01-01