系统和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发送消息的时候需要满足以下任一条件:
- 消息缓存大小达到batch.size。
- 缓存中的消息存时间在超过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
读提交,只有事务提交了才能消费到。防止了数据重复。
网友评论