美文网首页Java玩转大数据
Kafka场景性能优化配置

Kafka场景性能优化配置

作者: AlienPaul | 来源:发表于2024-12-02 13:46 被阅读0次

系统和JVM级别优化

  • 降低vm.swappiness。让系统更不倾向于使用swap空间。提高响应速度。
  • 提高ulimit -n(打开文件数)限制。
  • 提高vm.max_map_count限制。
  • 提高net.core.somaxconn连接数限制。
  • 提高net.ipv4.tcp_max_syn_backlog半连接数限制。
  • 使用G1垃圾回收,配置堆内存上限为6-8G以上。
  • 如果GC日志遇到大量分配大对象的信息,提高-XX:+G1HeapRegionSize=N配置值,加大region大小。

Kafka通用参数

未分类配置:

  • num.partitions: 默认分区数
  • default.replication.factor: 默认副本数
  • log.dirs配置多个Kafka数据存放目录。这些目录挂载到不同的磁盘上,可最大化磁盘IO性能。
  • auto.create.topics.enable是否允许自动创建topic。实际使用时建议关闭。防止不小心配置错topic名称创建出大量无用的topic。也能够避免自动创建的topic使用默认的参数配置(partition和replicas数量),达不到最佳性能。

leader选举:

  • unclean.leader.election.enable是否允许不在ISR只能的副本选举成为leader。启用的话会增加topic的可用性但是会有潜在丢失数据的风险。
  • auto.leader.rebalance.enable是否允许topic leader定期重选举。检查时间周期为leader.imbalance.check.interval.seconds。如果leader不均衡超过leader.imbalance.per.broker.percentage,会触发leader重选举,通常会更换leader,代价较大会引发性能抖动。生产环境可考虑关闭此功能。

日志保留:

  • log.retention.hours/minutes/ms消息的保存时间。减少时间可降低存储容量需求。但是若将Kafka当作存储使用,酌情增大该值。
  • log.retention.bytes保留消息的总大小上限。默认为-1没有限制。
  • message.max.bytes允许的最大消息大小,建议和socket.request.max.bytes同步修改。提高上限可允许大消息。

主从同步:

  • num.replica.fetchers副本拉取数据线程数。默认是1。增加可加快副本追上主分区的速度,但是为增加主分区节点和网络的压力。
  • replica.fetch.min.bytes每批同步消息大小下限。增加该值可减少通信频次,提高吞吐量。但是会增大副本同步延迟。
  • replica.fetch.max.bytes每批同步消息大小上限。
  • replica.fetch.wait.max.ms follower拉取数据最大等待时间。控制副本同步延迟的上限。

写入磁盘配置:

  • log.flush.interval.messages log分区数据条数到达该值时,必须刷写磁盘。默认值是一个非常大的数,不会强制刷写磁盘。
  • log.flush.interval.ms刷写磁盘时间间隔。

ack:

  • 0不确认写入是否成功,吞吐量最高,信息可能会丢失。
  • 1表示leader数据落盘确认写入成功。可靠性比0好。吞吐量降低。但是若leader故障,写入的信息没来得及同步到follower,信息仍可能丢失。
  • all表示必须leader的所有的ISR都落盘才确认写入成功。吞吐量最低但可以最大程度确保数据丢失(所有Kafka节点故障时数据仍会丢失)。

批量:
linger.ms和batch.size两个参数共同控制kafka发送消息的吞吐量和延迟。
Kafka producer发送消息的时候需要满足以下任一条件:

  1. 消息缓存大小达到batch.size。
  2. 缓存中的消息存时间在超过linger.ms,即便消息大小达不到batch.size,也会一并发送。

Consumer还有一个配置为max.poll.records,含义是每次拉取的最大消息数量。需要根据Consumer机器内存大小,业务复杂度和数据量调整为一个合适的值。

压缩:
compression.type指定压缩算法。通过CPU换取吞吐量。

消费者超时/再平衡配置:

  • session.timeout.ms:Consumer会话超时的时间。默认值为10秒。如果这个时间限制内coordinator如果没有接收到consumer的心跳,会认为consumer已故障,踢出consumer group。也就是说检测出consumer故障的最长耗时为该配置值。
  • heartbeat.interval.ms:Consumer向group coordinator发送心跳的间隔时间。
  • max.poll.interval.ms:Consumer拉取数据间隔时间上限。如果两次拉取时间间隔超过该上限consumer会被踢出consumer group。

频繁的超时会触发大量消费者rebalance,对性能有严重影响。需要根据业务处理压力和网络情况,调整这三个参数。在consumer压力大响应时间长的情况下,不合理的参数会导致某consumer反复加入然后被踢出消费者组,严重影响性能。使用者需要确保不要误触发consumer再平衡机制。

Broker网络:

  • num.network.threads网络通信线程数。
  • num.io.threads IO线程数。
  • socket.send.buffer.bytes网络发送缓冲区大小。
  • socket.receive.buffer.bytes网络接收缓冲区大小。
  • socket.request.max.bytes请求大小上限,提高上限可允许单条较大的消息发送和接收。

消息大小限制:

  • Broker的message.max.bytes。Broker允许的最大批量大小。间接限制了最大消息大小。确保该参数值一定要大于producer的batch.size。
  • Topic的max.message.bytes。同上,但作用范围是topic。
  • buffer.memory producer的发送消息缓存大小。
  • max.request.size producer的发送消息大小限制。
  • 消费者的fetch.min.bytes,fetch.max.bytes,max.partition.fetch.bytes。均为批量大小,限制不到单条消息大小。增大这些值可以提高每次poll到的record数量。

场景优化

提高吞吐量

Java API Kafka Producer是线程安全,Kafka Consumer不是线程安全。

Broker:

  • 增加num.replica.fetchers值,不要超过CPU核心数。
  • GC参数,避免频繁发生Full GC。

Producer:

  • 适当增加batch.size参数。例如从默认的16KB增大到512KB或者1MB。
  • 适当增加linger.ms参数。例如10-100。
  • 配置compression.type=lz4或者zstd等,启用压缩。
  • 设置acks=0或者1.
  • 减少retries。
  • 如果多个线程共享同一个Producer实例,增加buffer.memory。

Consumer:

  • 使用多Consumer进程或者线程同时消费数据。
  • 增加fetch.min.bytes但数值。例如1KB或者更大的值。

降低延迟

Broker:

  • 适当增加num.replica.fetchers的值。

Producer:

  • 设置linger.ms=0。
  • 不启用压缩,设置compression.type=none。
  • 设置acks=1。

Consumer端:

  • 设置fetch.min.bytes=1。

精准一次投送

精准一次投送即数据不丢失不重复。

务必确保生产消费端开启事务和幂等性。

Producer配置:

  • 提高retries尝试重传次数。需要网络波动时数据被成功发送出去的机会更大。
  • ack配置为all/-1。ISR中所有副本写入成功之后才会确认。减少数据丢失的可能性。
  • enable.idempotence开启幂等性。Retry的时候确保数据不会重复。但是幂等性只能够保证分区级别的数据不重复。
  • 开启事务,配置transaction.id。另外需要修改程序逻辑,发送数据前开启事务,确保数据发送成功之后提交事务。从而确保事务内发送的数据的原子性。事务能够在幂等性的基础之上,做到跨分区甚至跨session的数据原子性提交。
  • max.request.size不能过小,否则大消息发送会失败,引起数据丢失。

另外,Producer调用API send的时候应使用send(ProducerRecord<K,V> record, Callback callback);。通过回调可知数据是否真正发送成功,从而有机会补救处理。常用的send(ProducerRecord<K,V> record);方法未异步发送,无论数据发送成功与否均会返回,吞吐量很高但是结果不可控。

Broker配置:

  • unclean.leader.election.enable配置为false。禁止非ISR副本被选举为leader,宁愿牺牲可用性也不能丢失数据。
  • 加大replication.factor,至少要配置为3。越多的副本意味着越小的丢失概率。
  • min.insync.replicas最少需要写入多少个副本才认为写入成功。需要配合ack=all/-1场景使用。ack=all并不意为着数据绝不丢失。一个极端的场景是ISR只有一个分区。ack=all/-1仅要求所有的ISR写入成功。此时如果该分区所在broker崩溃,数据依然丢失。min.insync.replicas参数可以保证至少写入n个副本成功才算是数据写入成功。配置为大于1的值可以显著降低丢失数据的风险。当然该配置值也不能设置过大,过大的值会导致topic大概率写入不成功。例如3节点集群配置min.insync.replicas为3,只要一个broker崩溃,topic就再也无法写入数据。

Consumer配置:

  • enable.auto.commit配置为false,禁用自动提交offset。何时提交offset由业务逻辑确定,在确保数据成功消费处理之后再提交offset。
  • 事务隔离级别isolation.level默认为read_uncommitted读未提交,可以消费到未提交事务的数据。需要修改为read_committed读提交,只有事务提交了才能消费到。防止了数据重复。

相关文章

  • Kafka 最佳实践

    Kafka 基本配置及性能优化 这里主要是 Kafka 集群基本配置的相关内容。 硬件要求 Kafka 集群基本硬...

  • Fiddler

    使用场景开发环境host配置前后端接口调试线上bugfix性能分析和优化

  • 美团关于大数据技术的文章

    Spark性能优化指南——基础篇Spark性能优化指南——高级篇Spark在美团的实践Kafka文件存储机制那些事...

  • kafka性能优化

    1.partition数量配置 partition数量由topic的并发决定,并发少则1个分区就可以,并发越高,分...

  • JAVA-每日一面 2022-01-14

    问:高并发系统如何做性能优化?如何防止库存超卖?答: 高并发系统性能优化:优化程序,优化服务配置,优化系统配置1....

  • Laravel性能优化整理

    Laravel性能优化 配置各种缓存

  • 90 kafka

    1,kafka在大数据的应用场景;2,kafka为什么性能非常高2,kafka常见名词和概念4,Partition...

  • A(19)Kafka参数优化

    19 Kafka参数优化 1)Broker参数配置(server.properties)1、网络和io操作线程配置...

  • 优化概要

    数据库的性能取决于很多因素,比如:表,查询,和配置。配置就是最简单的算法,用于调整程序逻辑,以便适应更多场景。优化...

  • mysql性能优化-慢查询分析、优化索引和配置

    mysql性能优化-慢查询分析、优化索引和配置 分类:Mysql/postgreSQL 目录 一、优化概述 二、查...

网友评论

    本文标题:Kafka场景性能优化配置

    本文链接:https://www.haomeiwen.com/subject/vixesjtx.html