写点什么

5 种 kafka 消费端性能优化方法

  • 2022 年 8 月 04 日
  • 本文字数:2517 字

    阅读完需:约 8 分钟

5种kafka消费端性能优化方法

本文分享自华为云社区《FusionInsight HD&MRS:kafka消费端性能优化方法》,作者: 穿夹克的坏猴子 。


kafka 消费端性能优化主要从下面几个方面优化:

1. 接口使用方面优化:


  • 旧版本 highlevel-consumer:偏移量信息存储在 zookeeper,最大消费线程数与分区数量相同,不推荐

  • 旧版本 simpleconsumer:自行选择存储偏移量的方式,可以实现多线程消费单分区,若无特殊的性能要求,不推荐

  • 新版本 highlevel-consumer:偏移量信息存储在 kafka 指定的 topic 中,默认情况下最大消费线程数与分区数量相同,可以实现多线程消费单分区,推荐

2. 参数调优(以下参数需根据现网环境评估调至合适的值):

2.1 旧版本消费者(kafka old API)参数调优


fetch.message.max.bytes:该参数为一次性从 kafka 集群中获取的数据块大小。在升级到 651 版本后这个参数需要调大,否则容易出现获取数据限制的报错。建议调整大小不小于 kafka 的服务端参数 message.max.bytes。


注意如何确认为旧版本:如果生产者的配置方式包含如下这些配置,则为旧版本:group.id/zookeeper.connect

2.2 新版本参数(kafka new API)参数调优


max.poll.records:意味消费者一次 poll()操作,能够获取的最大数据量,调整这个值能提升吞吐量,于此同时也需要同步提升 max.poll.interval.ms 的参数大小。


fetch.max.bytes:意味 server 端可返回给 consumer 的最大数据大小,增加可以提升吞吐量,但是在客户端和服务端网络延迟比较大的环境下,建议可以减小该值,防止业务处理数据超时。


heartbeat.interval.ms:消费超时时间,consumer 与 kafka 之间的超时时间,该参数不能超过 session.timeout.ms,通常设置为 session.timeout.ms 的三分之一,默认值:3000。


max.partition.fetch.bytes:限制每个 consumer 发起 fetch 请求时候,读到数据(record)的限制,设置过大,consumer 本地缓存的数据就会越多,可能影响内存的使用,默认值:1048576。


fetch.max.bytes:server 端可返回给 consumer 的最大数据大小,数值可大于 max.partition.fetch.bytes,一般设置为默认值即可,默认值:52428800

session.timeout.ms:使用 consumer 组管理 offset 时,consumer 与 broker 之间的心跳超时时间,如果 consumer 消费数据的频率非常低,建议增大这个参数值,默认值:10000。


auto.offset.reset:消费过程中无法找到数据消费到的 offset 位置,所选择的消费策略,earliest:从头开始消费,可能会消费到重复数据,latest:从数据末尾开始消费,可能会丢失数据。默认值:earlist。


max.poll.interval.ms:消费者在每一轮 poll() (拉取数据之间的最大时间延迟),如果此超时时间期满之前 poll()没有被再次调用,则消费者被视为失败,并且分组将触发 rebalance,以便将分区重新分配给别的成员。


如果,再两次 poll 之间需要添加过多复杂的,耗时的逻辑,需要延长这个时间,默认值:300s

max.poll.records:消费者一次 poll()操作,能够获取的最大数据量,增加这个参数值,会增加一次性拉取数据的数据量,确保拉取数据的时间,至少在 max.poll.interval.ms 规定的范围之内,默认值:500

2.3 Simpleconsumer 参数调优


simpleconsumer 在初始化阶段需要传一个 fetchsize 的参数,比如:consumer=new SimpleConsumer(leaderBroker,a_port,100000,64*1024,clientName)中 64*1024,该参数表示 simpleconsumer 一次性获取的数据大小,如果该值过大则可能会导致 request 时间过长,使用过程中应该降低这个值,保证消费频率。


使用 SimpleConsumer 的核心需求是:多线程消费单个分区,以达到提升性能的要求,如果没有这样需求,不建议使用这个这种消费方式

3. 消费端频繁 rebalance 导致性能下降调优:

3.1 因业务处理能力不足导致的:


session.timout.ms 控制心跳超时时间。


heartbeat.interval.ms 控制心跳发送频率,建议该值不超过 session.timout.ms 的三分之一。


max.poll.interval.ms 控制每次 poll 的间隔,时间=获取数据的时间+处理数据的时间,如果 max.poll.records 设定的值在 max.poll.interval.ms 指定的时间内没有处理完成会触发 rebalance,这里给出一个相对较为合理的配置,建议在预计的处理时间的基础上再加 1 分钟。


max.poll.records 每个批次处理的数据条数,默认为 500 条。如果处理能力较低,建议可以减小这个值。

3.2 非正常消费者频繁的访问 kafka 集群导致频繁 rebalance:


收集 kafka-request.log,查看异常的 topic 有哪些客户端节点在消费,cat kafka-request.* | grep “topic=topicName” | grep “apikey=FETCH” | awk –F’from connection’ ‘{print $2}’ | awk –F’;’ ‘{print $1}’ | awk –F’-’ ‘{print $2}’ | awk –F’:’ ‘{print $1}’ | sort | uniq –c | sort -nr ,找出不应该产生消费行为的节点,停止异常节点上消费者

4. 版本引发性能下降优化


FI 8.0.2 版本之前 kafka SimpleAclAuthorizer 鉴权异常导致性能下降,8.0.2 版本在使用非安全端口(21005 或者 9092 端口)时会出现集群性能下降的问题,表现:kafka-root.log 中出现大量 ExitcodeException:id:Default#Principal:no such user 报错。


解决办法:升级到 FI 8023 以上版本


临时规避办法:业务侧使用 21007 端口访问 kafka,去掉鉴权插件即 allow.everyone.if.no.acl.found=true,将以下 kafka 服务端配置置为空:authorizer.class.name=

5. FI 6513~6516 版本的内核问题引发的性能异常


6513 版本在 kafka 引入社区的的 lazy index 功能后,在新的 segment 创建的过程中可能会导致并发创建失败的问题,常见的报错(server.log 中)如以下两种类型:


  1. java.lang.InternalError: a fault occurred in a recent unsafe memory access operation in compiled Java code

  2. java.lang.IllegalArgumentException: requirement failed: Attempt to append to a full index


当出现以上两种类型的报错的时候可以断定是版本问题导致,问题预警如:https://support.huawei.com/enterprise/zh/bulletins-product/ENEWS2000007844


解决方案:升级到 6517 版本以上版本或者打入紧急补丁:https://support.huawei.com/enterprise/zh/cloud-computing/fusioninsight-hd-pid-21110924/software/251482609?idAbsPath=fixnode01%7C7919749%7C7941815%7C19942925%7C250430185%7C21110924


临时规避方案:重启异常的 broker 实例


点击关注,第一时间了解华为云新鲜技术~

发布于: 刚刚阅读数: 4
用户头像

提供全面深入的云计算技术干货 2020.07.14 加入

华为云开发者社区,提供全面深入的云计算前景分析、丰富的技术干货、程序样例,分享华为云前沿资讯动态,方便开发者快速成长与发展,欢迎提问、互动,多方位了解云计算! 传送门:https://bbs.huaweicloud.com/

评论

发布
暂无评论
5种kafka消费端性能优化方法_大数据_华为云开发者联盟_InfoQ写作社区