SpringBoot中@KafkaListener原理

一时失言乱红尘 2021-06-10 20:39 812阅读 0赞

## 先上流程图： ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xiaDE5OTQ2Ng_size_16_color_FFFFFF_t_70_pic_center]

在我们的SpringBoot工程中，只需在方法中注解  
@KafkaListener(topics = \{”demo\_topic\_01“\})  
即可实现对该topic的监听

我们知道，kafka的consumer端通过从broker poll消息，然后处理

我们通过正向理解 + 结果反推 来分析

代码最终肯定会调用KafkaConsumer的poll()方法，可以通过方法栈来确定调用来源

### 查看poll的调用路径 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xiaDE5OTQ2Ng_size_16_color_FFFFFF_t_70]  
由调用知，KafkaMessageListenerContainer的内部类ListenerConsumer 实现了Runnable接口，里面调用了consumer的poll方法

观察Async stacktrace可知调用的起始位置  
KafkaListenerEndpointRegistry的start()方法

@Override
    	public void start() {
    		for (MessageListenerContainer listenerContainer : getListenerContainers()) {
    			startIfNecessary(listenerContainer);
    		}
    	}

最终调到了  
KafkaMessageListenerContainer的doStart方法  
这个方法 创建了一个新的KafkaConsumer对象  
可以理解成一个topic的监听器 就会创建一个对应的KafkaConsumer对象，只监听对应的topic消息

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xiaDE5OTQ2Ng_size_16_color_FFFFFF_t_70 1]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xiaDE5OTQ2Ng_size_16_color_FFFFFF_t_70 2]

下面我们从正流程走，在KafkaListenerEndpointRegistry的start()方法处打断点  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xiaDE5OTQ2Ng_size_16_color_FFFFFF_t_70 3]  
list里面包含了这个工程定义的所有的监听topic listener

这样整个流程就串起来了

### @KafkaListener 到 KafkaConsumer 流程 ###

KafkaMessageListenerContainer中创建了一个线程，这个线程new一个KafkaConsumer对象并执行poll方法。

每个注解了@KafkaListener的方法都会在KafkaListenerAnnotationBeanPostProcessor中，随之Spring的启动，创建相应的KafkaMessageListenerContainer。

看到BeanPostProcessor我们应该很熟悉，Spring初始化的过程中，会执行所有实现了BeanPostProcessor接口的postProcessBeforeInitialization、postProcessAfterInitializationr方法

@KafkaListener注释：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xiaDE5OTQ2Ng_size_16_color_FFFFFF_t_70 4]  
KafkaListenerAnnotationBeanPostProcessor关键代码：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xiaDE5OTQ2Ng_size_16_color_FFFFFF_t_70 5]  
后置处理器中，默认情况下只是将注解中的相关信息放入了一个List中，真正创建MessageListenerContainer的地方是在KafkaListenerEndpointRegistry中  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xiaDE5OTQ2Ng_size_16_color_FFFFFF_t_70 6]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xiaDE5OTQ2Ng_size_16_color_FFFFFF_t_70 7]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xiaDE5OTQ2Ng_size_16_color_FFFFFF_t_70 8]

## 总结 ##

这样，由@KafkaListener开始 ->生成KafkaMessageListenerContainer->创建KafkaConsumer对象->拉取消息，就串起来了~

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xiaDE5OTQ2Ng_size_16_color_FFFFFF_t_70_pic_center]: /images/20210527/5ddc827db6ed4d92b5c69c2c6c232e3b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xiaDE5OTQ2Ng_size_16_color_FFFFFF_t_70]: /images/20210527/e919c88652b8427792cb34d1fd0f74a9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xiaDE5OTQ2Ng_size_16_color_FFFFFF_t_70 1]: /images/20210527/2c53671f712f4c02a11a9443868b2711.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xiaDE5OTQ2Ng_size_16_color_FFFFFF_t_70 2]: /images/20210527/a9714fdac7dc4cfe84cdf449256c3f41.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xiaDE5OTQ2Ng_size_16_color_FFFFFF_t_70 3]: /images/20210527/7abf4532df234b29b290bfce186ebe6d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xiaDE5OTQ2Ng_size_16_color_FFFFFF_t_70 4]: /images/20210527/b8cc015706554c409c4f255c9e6d468d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xiaDE5OTQ2Ng_size_16_color_FFFFFF_t_70 5]: /images/20210527/cd8177c9c8474739aa78980cd73af3b5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xiaDE5OTQ2Ng_size_16_color_FFFFFF_t_70 6]: /images/20210527/c9b71ce3e90b4e7f9260cc51f4ed6e28.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xiaDE5OTQ2Ng_size_16_color_FFFFFF_t_70 7]: /images/20210527/45a720d9a22b44d2b04b2ae257c7c63f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xiaDE5OTQ2Ng_size_16_color_FFFFFF_t_70 8]: /images/20210527/daaa1a1c187c4a66aec8f7454c25dfe6.png