KafKa -相关参数优化

一、Broker参数调优

1. 处理消息的最大线程数

broker 处理消息的最大线程数，默认为 3，建议设为 cpu 核数 + 1：

例如：cpu 核数 8 ：

num.network.threads = 9

2. 处理磁盘 IO 的线程数

broker 处理磁盘 IO 的线程数，建议设为 cpu 核数 x 2 ：

例如：cpu 核数 8：

num.io.threads = 16

3. 数据落盘策略

Kafka重度依赖底层操作系统提供的

PageCache

功能。当上层有写操作时，操作系统只是将数据写入

PageCache

，同时标记

Page

属性为

Dirty

。当读操作发生时，先从

PageCache

中查找，如果发生缺页才进行磁盘调度，最终返回需要的数据。实际上

PageCache

是把尽可能多的空闲内存都当做了磁盘缓存来使用。但是也带来了问题，如果此时操作系统挂了数据就会丢失，可以通过时间间隔核消息的数量进行合理设置：

##每当producer写入10000条消息时，刷数据到磁盘
log.flush.interval.messages=10000

##每间隔5秒钟时间，刷数据到磁盘
log.flush.interval.ms=5000

4. segment 分段存储策略

分段文件配置默认是500mb ，有利于快速回收磁盘空间，重启kafka加载也会加快(如果文件过小，则文件数量比较多，kafka启动时是单线程扫描目录(log.dir)下所有数据文件)，文件较多时性能会稍微降低。

##日志滚动的周期时间，到达指定周期时间时，强制生成一个新的segment
log.roll.hours=72

## segment的索引文件最大尺寸限制，即时log.segment.bytes没达到，也会生成一个新的segment
log.index.size.max.bytes=10*1024*1024##控制日志segment文件的大小，超出该大小则追加到一个新的日志segment文件中（-1表示没有限制）
log.segment.bytes=1024*1024*1024

5. 日志清理策略

kafka 的消息不管是消费过还是没有消费，都会持久化到硬盘中，如果没有良好的日志清理策略，久而久之会占满磁盘空间，同样核上面配置相似，可以根据时间间隔和日志文件的大小来定义：

##   开启日志清理
log.cleaner.enable=true

##  日志清理运行的线程数
log.cleaner.threads = 2

##  日志清理策略选择有：delete和compact主要针对过期数据的处理，或是日志文件达到限制的额度，会被 topic创建时的指定参数覆盖，默认 delete
log.cleanup.policy = delete

##  数据文件保留多长时间， 存储的最大时间超过这个时间会根据log.cleanup.policy设置数据清除策略##  log.retention.bytes和 log.retention.minutes或 log.retention.hours任意一个达到要求，都会执行删除## 300 分钟
log.retention.minutes=300
## 24小时
log.retention.hours=24

##   topic每个分区的最大文件大小，-1没有大小限
log.retention.bytes=1G

##  文件大小检查的周期时间，是否触发 log.cleanup.policy中设置的策略
log.retention.check.interval.ms=5minutes

6. 基础配置

## 是否允许自动创建topic，若是false，就需要通过命令创建topic
auto.create.topics.enable =true

## 默认副本的数量，可以根据 Broker 的个数进行设置。
default.replication.factor = 3

## 默认，每个topic的分区个数，若是在topic创建时候没有指定的话会被topic创建时的指定参数覆盖
num.partitions = 3

## 消息体的最大大小，单位是字节，如果发送的消息过大，可以适当的增大该参数
message.max.bytes = 6525000

## socket的发送缓冲区的大小
socket.send.buffer.bytes=102400

## socket的接受缓冲区的大小
socket.request.max.bytes=104857600

7. 副本同步策略

## 默认10s，isr中的follow没有向isr发送心跳包就会被移除
replica.lag.time.max.ms = 10000

## 根据leader 和副本的信息条数差值决定是否从isr 中剔除此副本，此信息条数差值根据配置参数，在broker数量较少,或者网络不足的环境中,建议提高此值.
replica.lag.max.messages = 4000

## follower与leader之间的socket超时时间
replica.socket.timeout.ms=30*1000## 数据同步时的socket缓存大小
replica.socket.receive.buffer.bytes=64*1024## replicas每次获取数据的最大大小
replica.fetch.max.bytes =1024*1024## replicas同leader之间通信的最大等待时间，失败了会重试
replica.fetch.wait.max.ms =500

## fetch的最小数据尺寸，如果leader中尚未同步的数据不足此值,将会阻塞,直到满足条件
replica.fetch.min.bytes =1

## leader进行复制的线程数，增大这个数值会增加follower的IO
num.replica.fetchers=1

## 每个replica检查是否将最高水位进行固化的频率
replica.high.watermark.checkpoint.interval.ms = 5000

## leader的不平衡比例，若是超过这个数值，会对分区进行重新的平衡
leader.imbalance.per.broker.percentage = 10

## 检查leader是否不平衡的时间间隔
leader.imbalance.check.interval.seconds = 300

二、producer参数

spring:kafka:# kafka服务器地址(可以多个)bootstrap-servers: 10.218.222.39:9092,10.218.222.40:9092,10.218.222.41:9092producer:# key/value的序列化key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      # 批量抓取发送的的缓存大小，默认是16kB，意思是缓存中的数据达到配置的数值大小，kafka的生产端发送数据batch-size:65536# 缓存容量 默认值为33554432# 用于指定producer端用于缓存消息的缓冲区大小，单位为字节，默认值为：33554432  32M。# kafka采用的是异步发送的消息架构，producer启动时会首先创建一块内存缓冲区用于保存待发送的消息，# 然后由一个专属线程负责从缓冲区读取消息进行真正的发送。# 消息持续发送过程中，当缓冲区被填满后，producer立即进入阻塞状态直到空闲内存被释放出来，这段时间不能超过max.blocks.ms设置的值，# 一旦超过，producer则会抛出TimeoutException 异常，因为Producer是线程安全的，若一直报TimeoutException，需要考虑调高buffer.memory了。# 用户在使用多个线程共享kafka producer时，很容易把 buffer.memory 打满。buffer-memory:524288#失败重试次数retries:3# ACK# 0：Producer 不等待 Broker 的 ACK，这提供了最低延迟，Broker 一收到数据还没有写入磁盘就已经返回，当 Broker 故障时有可能丢失数据。# 1：Producer 等待 Broker 的 ACK，Partition 的 Leader 落盘成功后返回 ACK，如果在 Follower 同步成功之前 Leader 故障，那么将会丢失数据。# -1（all）：Producer 等待 Broker 的 ACK，Partition 的 Leader 和 Follower 全部落盘成功后才返回 ACK。但是在 Broker 发送 ACK 时，Leader 发生故障，则会造成数据重复。acks:1

三、consumer参数

spring:kafka:bootstrap-servers: 192.168.10.1:9092,192.168.10.2:9092,192.168.10.3:9092consumer:#用于标识此使用者所属的使用者组的唯一字符串。group-id: consumer
      #当Kafka中没有初始偏移量或者服务器上不再存在当前偏移量时该怎么办，默认值为latest，表示自动将偏移重置为最新的偏移量#可选的值为latest, earliest, noneauto-offset-reset: earliest
      #如果'enable.auto.commit'为true，则消费者偏移自动提交给Kafka的频率（以毫秒为单位），默认值为5000。auto-commit-interval: 5000ms
      #如果为true，则消费者的偏移量将在后台定期提交，默认值为trueenable-auto-commit:false#如果没有足够的数据立即满足“fetch.min.bytes”给出的要求，服务器在回答获取请求之前将阻塞的最长时间（以毫秒为单位）#默认值为500fetch-max-wait: 500ms
      #服务器应以字节为单位返回获取请求的最小数据量，默认值为1，对应的kafka的参数为fetch.min.bytes。fetch-min-size:1#心跳与消费者协调员之间的预期时间（以毫秒为单位），默认值为3000heartbeat-interval: 3000ms
      #密钥的反序列化器类，实现类实现了接口org.apache.kafka.common.serialization.Deserializerkey-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      #值的反序列化器类，实现类实现了接口org.apache.kafka.common.serialization.Deserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      #一次调用poll()操作时返回的最大记录数，默认值为500max-poll-records:50properties:max:poll:interval:ms:600000#0.10.1.0版本后新增的，这个参数需要根据实际业务处理时间进行设置，一旦Consumer处理不过来，就会被踢出Consumer Groupsession:timeout:ms:1800000listener:# 监听器的 AckMode# MANUAL :当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后, 手动调用Acknowledgment.acknowledge()后提交# MANUAL_IMMEDIATE : 手动调用Acknowledgment.acknowledge()后立即提交# RECORD : 当每一条记录被消费者监听器（ListenerConsumer）处理之后提交# BATCH :当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后提交# TIME 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，距离上次提交时间大于TIME时提交# COUNT 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，被处理record数量大于等于COUNT时提交ack-mode: manual
      ## poll拉取数据超时时间poll-timeout:600000# ack_mode为COUNT/COUNT_TIME 时配置ack-count:10# ack_mode为/COUNT_TIME 时配置ack-time:60000