@KafkaListener 使用方式
@KafkaListener 如何使用spring-kafka使用基于@KafkaListener注解,@KafkaListener使用方式如下@KafkaListener(topics = "topic1")public void kafkaListen(List<ConsumerRecord<xxx, xxx>> records) {...}在注解内指定topic名称,
@KafkaListener 如何使用
spring-kafka使用基于@KafkaListener注解,@KafkaListener使用方式如下
@KafkaListener(topics = "topic1")
public void kafkaListen(List<ConsumerRecord<xxx, xxx>> records) {
...
}
在注解内指定topic名称,当对应的topic内有新的消息时,testListen方法会被调用,参数就是topic内新的消息。这个过程是异步进行的。
@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注解标识的方法,将组织后的消息作为参数传入方法,执行用户逻辑。
@KafkaListener和@KafkaListners
@KafkaListeners是@KafkaListener的Container Annotation,这也是jdk8的新特性之一,注解可以重复标注。
@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) {}
更多推荐
所有评论(0)