Kafka 相关参数以及可靠性-Toy模板网

这篇具有很好参考价值的文章主要介绍了Kafka 相关参数以及可靠性。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

一、可靠性保证

1、消息存储可靠性

Kafka 通过持久化消息到磁盘来保障消息存储的可靠性，但是消息都是先写到操作系统的页缓存中，如果没有fsync到磁盘，存在消息丢失的可能性

Kafka 提供了两个参数来控制 Broker 的刷盘时机：

log.flush.interval.ms
long型，默认值null，单位ms，用于控制日志刷盘的时间间隔，每隔多少时间将消息刷到磁盘上

log.flush.interval.messages
long型，默认值9223372036854775807，用于控制日志刷盘的消息量，即每积累多少条消息将消息刷到磁盘上

建议配置：

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

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

频繁调用fsync会影响性能，官方不建议通过上述参数来强制写盘，认为数据的可靠性通过replica来保证，而强制flush数据到磁盘会对整体性能产生影响

2、broker 多副本可靠性

hw和isr

https://www.cnblogs.com/huxi2b/p/7453543.html

Kafka 水位详解_小小酥_LH的博客-CSDN博客

hw(水位)你可以理解成是一个全局(所有副本最小offset)的offset，针对的是一个分区

LEO代表着该副本的所有消息的最大offset，针对的是一个副本，也就是每个副本都有LEO，并且不一样。其中所有副本中最小的LEO就是水位

kafka不是完全同步，也不是完全异步，是一种ISR机制：
1. leader会维护一个与其基本保持同步的Replica列表，该列表称为ISR(in-sync Replica)，每个Partition都会有一个ISR，而且是由leader动态维护
2. 如果一个flower比一个leader落后太多，或者超过一定时间未发起数据复制请求，则leader将其重ISR中移除
3. 当ISR中所有Replica都向Leader发送ACK时，leader才commit

broker的topic一般多分区，每个分区保存在多个副本上，就算leader副本宕机也不会造成数据丢失。

其中包括了几个参数：replication.factor 和 min.insync.replicas、unclean.leader.election.enable

replication.factor指定Kafka副本数量，包含leader副本
min.insync.replicas该参数用于设定 ISR 中的最小副本数，默认值为 1，当且仅当（request.required.acks 参数设置为 -1 或者 all 时，此参数才生效。当 ISR 中的副本数少于 min.insync.replicas 配置的数量时，客户端会返回异常）
unclean.leader.election.enable=false 关闭unclean leader选举，即不允许非ISR中的副本被选举为leader，因为可能副本消息落后leader太多，以避免数据丢失

replication.factor >= 3 副本数量大于3，根据具体broker数量配置副本数量，尽量让副本在broker上均匀分配
min.insync.replicas > 1 该参数用于设定 ISR 中的最小副本数，默认值为 1
replication.factor > min.insync.replicas 如果两者相等，当一个副本挂掉了分区也就没法正常工作了，推荐设置replication.factor = min.insync.replicas + 1即可，即可以容忍一台Broker宕机
broker配置：

default.replication.factor=3
min.insync.replicas=2
replication.factor = min.insync.replicas + 1

3、消费端提交可靠性

Consumer中记录消费位移，注意，它记录的是要消费的下一条消息的位移，不是目前最新消息消费的位移

用户角度：

自动提交
手动提交

consumer角度：

同步提交
异步提交

1）自动提交
参数： enable.auto.commit为true 手动提交，默认方式，消费者每隔5秒提交上轮poll方法全部处理完的消息的位移。

假设在poll后3秒发生重平衡，消费者这时只提交了上一轮的位移，所以再次消费时会发生重复消费

2）手动提交
参数： enable.auto.commit为false 手动提交，一般推荐使用该方式。

手动同步提交：

先调用poll方法，立即使用commitSync()，如果在处理消息的时候发送异常或者再均衡，这期间的数据就丢失了
先调用poll方法，处理完数据再commitSync()，和上面自动提交一样，中间再均衡会重复消费
commitSync()是同步提交偏移量，主程序会一直阻塞，偏移量提交成功后才往下运行，这回限制Kafka的吞吐量。而降低提交频次又会造成重复消费的消息更多。

commitSync只要没有发生不可恢复错误，会进行重试，直到成功

commitAsync是异步提交，不会进行重试，如果重试，那么小偏移量会覆盖大偏移量，造成重复消费。举个例子，假如我们发起了一个异步提交commitA，此时的提交位移为2000，随后又发起了一个异步提交commitB且位移为3000；commitA提交失败但commitB提交成功，此时commitA进行重试并成功的话，会将实际上将已经提交的位移从3000回滚到2000，导致消息重复消费

4、生产者可靠性

1）acks机制

acks参数制定了必须要有多少个分区副本接收到消息，生产者才会认为消息写入是成功的。此参数对消息丢失的可能性有重要影响

request.required.acks = 0 消息的强制备份数量为 0，Producer 不停向 Leader 发送数据，而不需要 Leader 反馈成功消息，可能在发送过程中丢失数据，可能在 Leader 宕机时丢失数据，不推荐使用

request.required.acks = 1 默认情况，即消息的强制备份数量为 1，Producer 发送数据到 Leader，只要 Leader 成功写入本地日志成功，即返回客户端成功，不要求 ISR 中的其它副本与 Leader 保持同步

request.required.acks = -1（all）消息的强制备份数量为 ISR 列表中副本的数量，生产者写消息到leader副本中，必须要等leader副本数据完全同步到ISR中所有follower，消息才算commit

2）retries

生产者从服务器收到的错误有可能是临时性的错误(比如分区找不到leader)。这种情况下，retries参数的值决定了生产者可以重发消息的次数，若达到这个次数，生产者会放弃重试并返回错误。默认情况下，生产者会在每次重试之间等待100ms，也可以通过retry.backoff.ms参数来改变这个时间间隔。

3）同步和异步

同步发送消息时未同步到所有副本，leader replica所在broker宕机，消息丢失（不建议）
异步发送消息，新版本会将消息写入缓冲区，利于后台IO线程扫描缓冲区，将满足条件的消息封装并发送到broker，导致异常数据的两种情况：

消息量过大，导致缓冲区写满，此时缓冲区如果抛出异常或者数据被清除，都导致消息丢失
异步发送消息后使用回调函数，会进行重试，造成乱序

二、必要参数

参数说明

参数	默认值	取值范围	说明
batch.size	16K	>=0	单个parition对应的batch大小（bytes），缓存消息达到batch size之后，立即发送
linger.ms	1000ms	>=0	消息在batch中停留时间，达到linger.ms之后，立即发送
buffer.memory	32M	>=0	producer缓存消息，总的可用memory空间，如果memory用完，produder会立刻发送缓存中的消息
max.in.flight.requests.per.connnection			指定生产者在收到服务器响应之前可以发送多少个消息。值越高，就会占用越多的内存，不过也会提升吞吐量。设为1可以保证消息是按照发送的顺序写入服务器的，即使放生了重试。
timeout.ms、request.timeout.ms和metadata.fetch.timeout.ms			request.timeout.ms指定生产者在发送数据时等待服务器返回响应的时间，metadata.fetch.timeout.ms指定生产者在获取元数据(比如目标分区的leader是谁)时等待服务器返回响应时间。timeout.ms指定了broker等待同步副本返回消息确认的时间，与asks的配置相匹配——若指定时间内没收到同步副本的确认，则broker就会返回一个错误。
max.block.ms			指定在调用send()方法或使用partitionsFor()方法获取元数据时生产者的阻塞时间。当生产者的发送缓冲区已满或没有可用元数据时，这些方法就会阻塞。当阻塞时间达到max.block.ms时，生产者会抛出超时异常。