kafka消费端在springboot项目中简单注释使用方法

我会带着你远行 2021-09-23 05:14 658阅读 0赞

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

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

在注解内指定topic名称，当对应的topic有新消息到来时，testListen方法会被调用，参数就是topic内的新消息。这个过程是异步进行的。

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

1.解析：解析@KafkaListener注解。  
2.注册：将解析后的数据注册到spring-kafka。  
3.监听：开始监听topic消息。  
4.调用：调用注解标识的方法，将监听到的数据作为参数传入。  
下面，我们一步一步分析。

2.1解析  
@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注解。

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

2.3监听  
注册完成之后，每个Listener Container会开始工作，会新启一个新的线程，用于初始化KafkaConsumer、监听topic消息等。

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