首页 > 软件编程 > java > @KafkaListener 使用

@KafkaListener 如何使用

作者:DemonHunter211

这篇文章主要介绍了@KafkaListener 如何使用,本文通过图文实例代码相结合给大家详细讲解,文末给大家介绍了kafka的消费者分区分配策略,需要的朋友可以参考下

@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) {}

扩展:kafka的消费者分区分配策略

kafka有三种分区分配策略

1. RoundRobin

2. Range

3. Sticky

1. RoundRobin

(1)把所有topic的分区partition放入一个队列中,按照name的hashcode进行排序;

(2)把consumer放在一个循环队列,按照name的hashcode进行排序;

(3)循环遍历consumer,从partition队列pop出一个partition,分配给当前consumer;以此类推,取下一个consumer,继续从partition队列pop出来分配给当前consumer;直到partition队列中的元素被分配完;

2. Range

(1)假设topicA有4个分区,topicB有5个分区,topicC有6个分区;一共有3个consumer;

(2)遍历3个topic的分区集合,先取topicA的分区集合,然后准备依次给3个consumer分配分区;对于第1个consumer,所分配的分区数量根据以下公式:假设消费者数量为N,当前主题剩下的分区数量为M,则当前消费者应该分配的分区数量 = M%N==0? M/N +1 : M/N ;按照公式,3个消费者应该分配的分区数量依次为:2/1/1,即topicA-partition-0/1分配给consumer-0,topicA-partition-2分配给consumer-1,topicA-partition-3分配给consumer-2;

(3)按照上述规则按序把topicB和topicC的分区分配给3个consumer;依次为:2/2/1,2/2/2;

3. Sticky

kafka在0.11版本引入了Sticky分区分配策略,它的两个主要目的是:

1. 分区的分配要尽可能的均匀,分配给消费者者的主题分区数最多相差一个;

2. 分区的分配尽可能的与上次分配的保持相同;

当两者发生冲突时,第一个目标优先于第二个目标;

粘性分区是由Kafka从0.11x版本开始引入的分配策略,首先会尽量均衡的分配分区到消费者上面,在出现同一消费组内消费者出现问题的时候,会尽量保持原来的分配的分区不变;

Sticky分区初始分配分区的方法与Range相似,但是不同;拿7个分区3个消费者为例,消费者消费的分区依旧是3/2/2,但是不同与Range的是Range分区是排好序的,但是Sticky分区是随机的;

到此这篇关于@KafkaListener 如何使用方式的文章就介绍到这了,更多相关@KafkaListener 使用内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

您可能感兴趣的文章:
  • Java中Swagger框架的使用详解
    Java中Swagger框架的使用详解
    2023-08-08
  • mybatis查询SqlServer慢问题及解决
    mybatis查询SqlServer慢问题及解决
    2023-08-08
  • 使用Mock进行业务逻辑层Service测试详解
    使用Mock进行业务逻辑层Service测试详解
    2023-08-08
  • Java如何获取真实请求IP
    Java如何获取真实请求IP
    2023-08-08
  • 详解Java中NullPointerException的处理方法
    详解Java中NullPointerException的处理方法
    2023-08-08
  • IDEA maven上传速度很慢的解决办法
    IDEA maven上传速度很慢的解决办法
    2023-08-08
  • mybatis查询SqlServer慢问题及解决
    mybatis查询SqlServer慢问题及解决
    2023-08-08
  • Spring WebFlux使用函数式编程模型构建异步非阻塞服务
    Spring WebFlux使用函数式编程模型构建异步非阻塞服务
    2023-08-08
  • 美国设下计谋,用娘炮文化重塑日本,已影响至中国
    美国设下计谋,用娘炮文化重塑日本,已影响至中国
    2021-11-19
  • 时空伴随者是什么意思?时空伴随者介绍
    时空伴随者是什么意思?时空伴随者介绍
    2021-11-09
  • 工信部称网盘企业免费用户最低速率应满足基本下载需求,天翼云盘回应:坚决支持,始终
    工信部称网盘企业免费用户最低速率应满足基本下载需求,天翼云盘回应:坚决支持,始终
    2021-11-05
  • 2022年放假安排出炉:五一连休5天 2022年所有节日一览表
    2022年放假安排出炉:五一连休5天 2022年所有节日一览表
    2021-10-26
  • 电脑版 - 返回首页

    2006-2023 脚本之家 JB51.Net , All Rights Reserved.
    苏ICP备14036222号