备案控制台
开发者社区 云计算 文章 正文

【spring-kafka】属性concurrency的作用及如何配置(RoundRobinAssignor 、RangeAssignor)

2021-12-28 587
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 【spring-kafka】属性concurrency的作用及如何配置(RoundRobinAssignor 、RangeAssignor)

目录

concurrency属性作用

什么情况下设置concurrency,以及设置多少

RoundRobinAssignor 和 RangeAssignor 作用

不同配置的实验分析

分区数3|concurrency = 1|启动一个客户端(单机)

分区数3|concurrency = 1|启动2个客户端(分布式模式)

分区数3|concurrency = 3|启动一个客户端

分区数3|concurrency = 3|启动2个客户端(分布式模式)

批量消费

concurrency属性作用

concurrency默认是1;


container.setConcurrency(3)表示创建三个KafkaMessageListenerContainer实例。

一个KafkaMessageListenerContainer实例分配一个分区进行消费;

如果设置为1的情况下, 这一个实例消费Topic的所有分区;

如果设置多个,那么会平均分配所有分区;

如果实例>分区数; 那么空出来的实例会浪费掉;

如果实例<=分区数 那么会有一部分实例消费多个实例,但也是均衡分配的


如果在分布式情况下, 那么总的KafkaMessageListenerContainer实例数= 服务器机器数量*concurrency ;


什么情况下设置concurrency,以及设置多少

这个得看我们给Topic设置的分区数量; 总的来说就是 机器数量*concurrency <= 分区数


例如分区=3; 而且同时有3台机器 ,那么concurrency=1就行了; 设置多了就会浪费资源;、


例如分区=9; 只有3台机器;那么可以concurrency=3 ; 每台机器3个消费者连接3个分区; 那么你可能会问我们concurrency=1不也可以吗; 反正都是一台机器消费3个分区;

话是没有错; 但是他们的差别在 一个线程消费3个分区和 3个线程消费3个分区 , 单线程和多线程你选哪个


RoundRobinAssignor 和 RangeAssignor 作用

默认情况下 spring.kafka.consumer.properties.partition.assignment.strategy=\ org.apache.kafka.clients.consumer.RangeAssignor


假如如下情况,同时监听了2个Topic; 并且每个topic的分区都是3; concurrency设置为6;

    @KafkaListener(id = "consumer-id6", topics = {"SHI_TOPIC3","SHI_TOPIC4"}, containerFactory = "concurrencyFactory"
            , clientIdPrefix = "myClientId6")
    public void consumer6(List<?> list) {
        StringBuffer sb = new StringBuffer();
        list.forEach((l)->{
            sb.append("|msg:").append(l);
        });
        log.info("线程:{} consumer-id6 消费->{}",Thread.currentThread(),sb);
    }

那么你期望的是不是 2*3=6 刚好6个线程;一个线程分配一个分区; 那么我们运行看看结果

image.png

看上图中,我们发现并没有按照我们的预期去做; 有三个消费者其实是闲置状态的; 只有另外的3个消费者负责了2个Topic的总共6个分区; 因为默认的分配策略是 spring.kafka.consumer.properties.partition.assignment.strategy=\ org.apache.kafka.clients.consumer.RangeAssignor ;


如果想达到我们的预期;那你可以修改策略; spring.kafka.consumer.properties.partition.assignment.strategy=\ org.apache.kafka.clients.consumer.RoundRobinAssignor


修改之后

image.png

每个线程分配一个分区

不同配置的实验分析

分区数3|concurrency = 1|启动一个客户端(单机)

创建了名为 SHI_TOPIC3并且分区数为3的Topic

image.png

代码启动,设置concurrency = 1, 只启动一个客户端;

启动日志

2020-11-18 17:14:42 o.a.k.c.c.i.ConsumerCoordinator 611 [INFO] [Consumer clientId=myClientId5-0, groupId=consumer-id5] 
Finished assignment for group at generation 6: {myClientId5-0-a273480d-2370-49e5-9187-ed10fe6dcf51=
Assignment(partitions=[SHI_TOPIC3-0, SHI_TOPIC3-1, SHI_TOPIC3-2])}
 2020-11-18 17:14:42 o.s.k.l.KafkaMessageListenerContainer 292 [INFO] consumer-id5: 
 partitions assigned: [SHI_TOPIC3-2, SHI_TOPIC3-1, SHI_TOPIC3-0]

可以看到这个客户端myClientId5-0-a273480d-2370-49e5-9187-ed10fe6dcf51 被分配了3个分区SHI_TOPIC3-0, SHI_TOPIC3-1, SHI_TOPIC3-2;

消费日志

2020-11-18 17:14:45 c.d.b.k.KafkaListeners 109 [INFO] 线程:Thread[consumer-id5-0-C-1,5,main] consumer-id5 消费->record:ConsumerRecord(topic = SHI_TOPIC3, partition = 2, leaderEpoch = 0, offset = 0, CreateTime = 1605690882681, serialized key size = 13, serialized value size = 11, headers = RecordHeaders(headers = [], isReadOnly = false), key = 1605690882615, value = 我是data0),value:我是data0,partition:2,offset:0
 2020-11-18 17:14:45 c.d.b.k.KafkaListeners 109 [INFO] 线程:Thread[consumer-id5-0-C-1,5,main] consumer-id5 消费->record:ConsumerRecord(topic = SHI_TOPIC3, partition = 2, leaderEpoch = 0, offset = 1, CreateTime = 1605690882705, serialized key size = 13, serialized value size = 11, headers = RecordHeaders(headers = [], isReadOnly = false), key = 1605690882705, value = 我是data4),value:我是data4,partition:2,offset:1
 2020-11-18 17:14:45 c.d.b.k.KafkaListeners 109 [INFO] 线程:Thread[consumer-id5-0-C-1,5,main] consumer-id5 消费->record:ConsumerRecord(topic = SHI_TOPIC3, partition = 2, leaderEpoch = 0, offset = 2, CreateTime = 1605690882705, serialized key size = 13, serialized value size = 11, headers = RecordHeaders(headers = [], isReadOnly = false), key = 1605690882705, value = 我是data5),value:我是data5,partition:2,offset:2
 2020-11-18 17:14:45 c.d.b.k.KafkaListeners 109 [INFO] 线程:Thread[consumer-id5-0-C-1,5,main] consumer-id5 消费->record:ConsumerRecord(topic = SHI_TOPIC3, partition = 2, leaderEpoch = 0, offset = 3, CreateTime = 1605690882706, serialized key size = 13, serialized value size = 11, headers = RecordHeaders(headers = [], isReadOnly = false), key = 1605690882705, value = 我是data6),value:我是data6,partition:2,offset:3
 2020-11-18 17:14:45 c.d.b.k.KafkaListeners 109 [INFO] 线程:Thread[consumer-id5-0-C-1,5,main] consumer-id5 消费->record:ConsumerRecord(topic = SHI_TOPIC3, partition = 2, leaderEpoch = 0, offset = 4, CreateTime = 1605690882706, serialized key size = 13, serialized value size = 11, headers = RecordHeaders(headers = [], isReadOnly = false), key = 1605690882706, value = 我是data7),value:我是data7,partition:2,offset:4
.....

可以看到线程都是同一个 Thread[consumer-id5-0-C-1,5,main] ; 说明的问题就是 在消费的时候是单线程消费的,并且还是一个线程去消费 3个分区的数据; 又涉及到切换消费分区的问题;

查询这个消费组的消费情况;

image.png

也证实只有一个消费者myClientId5-0-a273480d-2370-49e5-9187-ed10fe6dcf51在消费3个分区的数据;


分区数3|concurrency = 1|启动2个客户端(分布式模式)

第一个客户端不动,继续运行, 然后启动第二个客户端

第一个客户端发生的变化

 2020-11-18 17:34:24 o.a.k.c.c.i.ConsumerCoordinator 611 [INFO] [Consumer clientId=myClientId5-0, groupId=consumer-id5] Finished assignment for group at generation 9: {myClientId5-0-66a81e88-d924-4890-8b8e-2c6960ed0704=Assignment(partitions=[SHI_TOPIC3-2]), myClientId5-0-31c9a99f-5735-4a1d-b537-95bc5ab4533f=Assignment(partitions=[SHI_TOPIC3-0, SHI_TOPIC3-1])}

第一个客户端进行了 再平衡 ; 因为多了第二个可以分担压力进行消费; 可以看到把SHI_TOPIC3-2平衡出去了

第二个客户端的日志

 2020-11-18 17:34:24 o.a.k.c.Metadata 277 [INFO] [Consumer clientId=myClientId5-0, groupId=consumer-id5] Cluster ID: O304VSOeSEyporzbs5AITA
 2020-11-18 17:34:24 o.a.k.c.c.i.AbstractCoordinator 797 [INFO] [Consumer clientId=myClientId5-0, groupId=consumer-id5] Discovered group coordinator xxxxxx:9092 (id: 2147483645 rack: null)
 2020-11-18 17:34:24 o.a.k.c.c.i.AbstractCoordinator 552 [INFO] [Consumer clientId=myClientId5-0, groupId=consumer-id5] (Re-)joining group
 2020-11-18 17:34:25 o.s.k.l.KafkaMessageListenerContainer 292 [INFO] consumer-id5: partitions assigned: [SHI_TOPIC3-2]

查询客户端消费情况

image.png

可以看到第二个客户端分配到了SHI_TOPIC3--2的分区进行消费; 并且是单线程消费;

分区数3|concurrency = 3|启动一个客户端

客户端日志

2020-11-18 17:50:42 o.a.k.c.c.i.ConsumerCoordinator 273 [INFO] [Consumer clientId=myClientId5-1, groupId=consumer-id5] Adding newly assigned partitions: SHI_TOPIC3-1
 2020-11-18 17:50:42 o.a.k.c.c.i.ConsumerCoordinator 273 [INFO] [Consumer clientId=myClientId5-0, groupId=consumer-id5] Adding newly assigned partitions: SHI_TOPIC3-0
 2020-11-18 17:50:42 o.a.k.c.c.i.ConsumerCoordinator 273 [INFO] [Consumer clientId=myClientId5-2, groupId=consumer-id5] Adding newly assigned partitions: SHI_TOPIC3-2
 2020-11-18 17:50:42 o.s.k.l.KafkaMessageListenerContainer 292 [INFO] consumer-id5: partitions assigned: [SHI_TOPIC3-2]
 2020-11-18 17:50:42 o.s.k.l.KafkaMessageListenerContainer 292 [INFO] consumer-id5: partitions assigned: [SHI_TOPIC3-0]
 2020-11-18 17:50:42 o.s.k.l.KafkaMessageListenerContainer 292 [INFO] consumer-id5: partitions assigned: [SHI_TOPIC3-1]

上面日志显示 创建了3个消费者,他们都属于同一个消费组groupId=consumer-id5,3个分区刚好3个消费者一人一个分区平均分配;

客户端日志

 2020-11-18 17:50:45 c.d.b.k.KafkaListeners 109 [INFO] 线程:Thread[consumer-id5-0-C-1,5,main] consumer-id5 消费->record:ConsumerRecord(topic = SHI_TOPIC3, partition = 0, leaderEpoch = 0, offset = 11, CreateTime = 1605693042720, serialized key size = 13, serialized value size = 11, headers = RecordHeaders(headers = [], isReadOnly = false), key = 1605693042432, value = 我是data0),value:我是data0,partition:0,offset:11
 2020-11-18 17:50:45 c.d.b.k.KafkaListeners 109 [INFO] 线程:Thread[consumer-id5-2-C-1,5,main] consumer-id5 消费->record:ConsumerRecord(topic = SHI_TOPIC3, partition = 2, leaderEpoch = 0, offset = 12, CreateTime = 1605693042751, serialized key size = 13, serialized value size = 11, headers = RecordHeaders(headers = [], isReadOnly = false), key = 1605693042750, value = 我是data1),value:我是data1,partition:2,offset:12
 2020-11-18 17:50:45 c.d.b.k.KafkaListeners 109 [INFO] 线程:Thread[consumer-id5-1-C-1,5,main] consumer-id5 消费->record:ConsumerRecord(topic = SHI_TOPIC3, partition = 1, leaderEpoch = 0, offset = 17, CreateTime = 1605693042757, serialized key size = 13, serialized value size = 11, headers = RecordHeaders(headers = [], isReadOnly = false), key = 1605693042757, value = 我是data7),value:我是data7,partition:1,offset:17

image.png

每个消费者都是单线程,一个线程消费一个分区

分区数3|concurrency = 3|启动2个客户端(分布式模式)

启动第一个客户端

image.png

启动第二个客户端

image.png

启动第二个客户端之后就发生了 再分配rebalance; 可以看到,总共就有6个消费者, 但是其中的3个都是处于空闲状态;
因为一个分区最多只能有一个分区来进行消费;

批量消费

  /**
     * 监听器工厂 批量消费
     * @return
     */
    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> concurrencyFactory() {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(kafkaConsumerFactory());
        factory.setConcurrency(1);
        //设置为批量消费，每个批次数量在Kafka配置参数中设置ConsumerConfig.MAX_POLL_RECORDS_CONFIG
        factory.setBatchListener(true);
        return factory;
    }

配置文件设置 批量的最大条数

kafka.consumer.max-poll-records = 20

消费

    @KafkaListener(id = "consumer-id6", topics = "SHI_TOPIC3", containerFactory = "concurrencyFactory"
            , clientIdPrefix = "myClientId6")
    public void consumer6(List<?> list) {
        StringBuffer sb = new StringBuffer();
        list.forEach((l)->{
            sb.append("|msg:").append(l);
        });
        log.info("线程:{} consumer-id6 消费->{}",Thread.currentThread(),sb);
    }



文章标签：
云消息队列 Kafka 版
日志服务
关键词：
Spring配置
Spring属性
云消息队列 Kafka 版配置
云消息队列 Kafka 版属性
石臻臻的杂货铺
目录
相关文章
wljslmz
|
19天前
|
存储 Java 数据安全/隐私保护
解放配置之道：Spring引入外部属性文件
【4月更文挑战第20天】
wljslmz
28 0
喜欢猪猪
|
4天前
|
安全 Java 开发者
深入理解Spring Boot配置绑定及其实战应用
【4月更文挑战第10天】本文详细探讨了Spring Boot中配置绑定的核心概念，并结合实战示例，展示了如何在项目中有效地使用这些技术来管理和绑定配置属性。
喜欢猪猪
11 1
砖业洋__
|
2天前
|
XML Java 数据格式
Spring高手之路18——从XML配置角度理解Spring AOP
本文是全面解析面向切面编程的实践指南。通过深入讲解切面、连接点、通知等关键概念，以及通过XML配置实现Spring AOP的步骤。
砖业洋__
21 6
Spring高手之路18——从XML配置角度理解Spring AOP
源码星辰
|
3天前
|
消息中间件 开发框架 Java
什么是Spring Boot 自动配置？
Spring Boot 是一个流行的 Java 开发框架，它提供了许多便利的功能和工具，帮助开发者快速构建应用程序。其中一个最引人注目的特性是其强大的自动配置功能。
源码星辰
7 0
qiuqiushuibx
|
6天前
|
Java Spring
Spring文件配置以及获取
Spring文件配置以及获取
qiuqiushuibx
13 0
xiaowang_lj
|
8天前
|
XML Java 数据格式
Spring 属性注入方式
Spring 属性注入方式
xiaowang_lj
13 2
码农烧烤
|
8天前
|
Java 数据库连接 数据库
Spring事务简介，事务角色，事务属性
Spring事务简介，事务角色，事务属性
码农烧烤
18 2
东方睿赢
|
9天前
|
Java Apache Spring
Spring BeanUtils与Apache BeanUtils提供基本属性复制，适用于简单需求
【5月更文挑战第4天】Spring BeanUtils与Apache BeanUtils提供基本属性复制，适用于简单需求；Cglib BeanCopier用于转换为Cglib代理对象；Apache PropertyUtils处理属性操作；Dozer支持复杂对象映射。选择工具取决于具体需求，如需精细控制或对象映射，推荐Dozer或Apache PropertyUtils。Apache BeanUtils可能因潜在的封装性破坏被禁用。
东方睿赢
22 3
Yeats_Liao
|
12天前
|
Java 微服务 Spring
Spring Boot中获取配置参数的几种方法
Spring Boot中获取配置参数的几种方法
Yeats_Liao
22 2
vohelon
|
14天前
|
消息中间件 安全 Java
在Spring Bean中，如何通过Java配置类定义Bean?
【4月更文挑战第30天】在Spring Bean中，如何通过Java配置类定义Bean?
vohelon
21 1

热门文章

最新文章

  • 1
    SpringBoot开发案例之整合Kafka实现消息队列
  • 2
    解决kafka集群由于默认的__consumer_offsets这个topic的默认的副本数为1而存在的单点故障问题
  • 3
    flink kafka connector源码解读（超详细）
  • 4
    Kafka实战－KafkaOffsetMonitor
  • 5
    《Kafka官方文档》主页
  • 6
    Flink之DataStream API(执行环境、数据源、读取kafka)
  • 7
    Kafka项目实战－用户日志上报实时统计之编码实践
  • 8
    05【在线日志分析】之Kafka 0.10.1.0 Cluster的搭建和Topic简单操作实验
  • 9
    linux如何安装KafKa
  • 10
    Kafka中的Controller（控制器）节点
  • 1
    【Springcloud Alibaba微服务分布式架构 | Spring Cloud】之学习笔记（一）基础知识+各个组件介绍+聚合父工程创建
    139
  • 2
    spring cache整合redis实现springboot项目中的缓存功能
    50
  • 3
    Spring Boot整合Spring Security--学习笔记
    62
  • 4
    【spring（六）】WebSocket网络传输协议
    32
  • 5
    【spring（五）】SpringMvc总结 SSM整合流程
    34
  • 6
    【spring（四）】Spring事务管理和@Transactional注解
    42
  • 7
    Spring Boot 4.0：构建云原生Java应用的前沿工具
    72
  • 8
    Java 新手如何使用Spring MVC 中的查询字符串和查询参数
    23
  • 9
    Spring Boot的定时任务与异步任务
    56
  • 10
    通用Shell脚本执行Spring Boot项目Jar包
    34

    • 相关课程

    更多
  • 全面讲解Spring Cloud Alibaba技术栈（知识精讲+项目实战）第四阶段
  • 精通Spring Cloud Alibaba
  • 微服务框架 Spring Cloud 快速入门
  • Java Web开发系列课程 - Spring框架入门
  • Spring Boot 2.5.x开发实战
  • Spring Cloud微服务架构设计与开发实战

    • 相关电子书

    更多
  • 云栖社区特邀专家徐雷Java Spring Boot开发实战系列课程（第20讲）：经典面试题与阿里等名企内部招聘求职面试技巧
  • 微服务架构模式与原理Spring Cloud开发实战
  • 阿里特邀专家徐雷Java Spring Boot开发实战系列课程（第18讲）：制作Java Docker镜像与推送到DockerHub和阿里云Docker仓库

    • 相关实验场景

    更多
  • Spring-cloud-bus-rocketmq入门与实践
  • Spring-cloud-stream-binder-rocketmq入门与实践
  • Rocketmq-spring入门与实践
  • 从零搭建Spring Boot的Hello World
    • 下一篇
    部署LAMP环境（Alibaba Cloud Linux 3）