本文会介绍一下@KafkaListener的工作流程,但是不会详细深入。

<>@KafkaListener 如何使用

spring-kafka使用基于@KafkaListener注解,@KafkaListener使用方式如下
@KafkaListener(topics = "xxx") public void testListen(List<ConsumerRecord<xxx,
xxx>> records) { ... }
在注解内指定topic名称,当对应的topic内有新的消息时,testListen方法会被调用,参数就是topic内新的消息。这个过程是异步进行的。

<>工作流程

上面介绍了@KafkaListener的使用方式,到这里应该是很好奇它到底是如何工作的。

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

* 解析;解析@KafkaListener注解。
* 注册;解析后的数据注册到spring-kafka。
* 监听;开始监听topic变更。
* 调用;调用注解标识的方法,将监听到的数据作为参数传入。
下面我们一步一步分析

<>解析

@KafkaListener注解由KafkaListenerAnnotationBeanPostProcessor类解析,后者实现了
BeanPostProcessor接口,这个接口如下
public interface BeanPostProcessor { Object postProcessBeforeInitialization(
Object bean, String beanName) throws BeansException; Object
postProcessAfterInitialization(Object bean, String beanName) throws
BeansException; }
接口内部有2个方法,分别在bean初始化前后被调用。

KafkaListenerAnnotationBeanPostProcessor内会在postProcessAfterInitialization方法内解析
@KafkaListener注解。

<>注册

解析步骤里,我们可以获取到所有含有@KafkaListener注解的类,之后这些类的相关信息会被注册到
KafkaListenerEndpointRegistry内,包括注解所在的方法,当前的bean等。KafkaListenerEndpointRegistry
这个类内部会维护多个Listener Container,每一个@KafkaListener都会对应一个Listener Container。并且每个
Container对应一个线程。

<>监听

注册完成之后,每个Listener Container会开始工作,会新启一个新的线程,初始化KafkaConsumer,监听topic变更等。

<>调用

监听到数据之后,container会组织消息的格式,随后调用解析得到的@KafkaListener
注解标识的方法,将组织后的消息作为参数传入方法,执行用户逻辑。

<>总结

上面4个步骤就是@KafkaListener工作的核心流程。

技术
下载桌面版
GitHub
百度网盘(提取码:draw)
Gitee
云服务器优惠
阿里云优惠券
腾讯云优惠券
华为云优惠券
站点信息
问题反馈
邮箱:ixiaoyang8@qq.com
QQ群:766591547
关注微信