Kafka无消息丢失配置

Kafka到底会不会丢数据(data loss)? 一般不会，但有些状况下的确有可能会发生。下面的参数配置及Best practice列表能够较好地保证数据的持久性(固然是trade-off，牺牲了吞吐量)。笔者会在该列表以后对列表中的每一项进行讨论，有兴趣的同窗能够看下后面的分析。

1. block.on.buffer.full = true
2. acks = all
3. retries = MAX_VALUE
4. max.in.flight.requests.per.connection = 1
5. 使用KafkaProducer.send(record, callback)
6. callback逻辑中显式关闭producer：close(0) 
7. unclean.leader.election.enable=false
8. replication.factor = 3 
9. min.insync.replicas = 2
10. replication.factor > min.insync.replicas
11. enable.auto.commit=false
12. 消息处理完成以后再提交位移

给出列表以后，咱们从两个方面来探讨一下数据为何会丢失：

1. Producer端

　　目前比较新版本的Kafka正式替换了Scala版本的old producer，使用了由Java重写的producer。新版本的producer采用异步发送机制。KafkaProducer.send(ProducerRecord)方法仅仅是把这条消息放入一个缓存中(即RecordAccumulator，本质上使用了队列来缓存记录)，同时后台的IO线程会不断扫描该缓存区，将知足条件的消息封装到某个batch中而后发送出去。显然，这个过程当中就有一个数据丢失的窗口：若IO线程发送以前client端挂掉了，累积在accumulator中的数据的确有可能会丢失。

　　Producer的另外一个问题是消息的乱序问题。假设客户端代码依次执行下面的语句将两条消息发到相同的分区

producer.send(record1);
producer.send(record2);

若是此时因为某些缘由(好比瞬时的网络抖动)致使record1没有成功发送，同时Kafka又配置了重试机制和max.in.flight.requests.per.connection大于1(默认值是5，原本就是大于1的)，那么重试record1成功后，record1在分区中就在record2以后，从而形成消息的乱序。不少某些要求强顺序保证的场景是不容许出现这种状况的。

　　鉴于producer的这两个问题，咱们应该如何规避呢？？对于消息丢失的问题，很容易想到的一个方案就是：既然异步发送有可能丢失数据， 我改为同步发送总能够吧？好比这样：

producer.send(record).get();

这样固然是能够的，可是性能会不好，不建议这样使用。所以特地总结了一份配置列表。我的认为该配置清单应该可以比较好地规避producer端数据丢失状况的发生：(特此说明一下，软件配置的不少决策都是trade-off，下面的配置也不例外：应用了这些配置，你可能会发现你的producer/consumer 吞吐量会降低，这是正常的，由于你换取了更高的数据安全性)

• block.on.buffer.full = true  尽管该参数在0.9.0.0已经被标记为“deprecated”，但鉴于它的含义很是直观，因此这里仍是显式设置它为true，使得producer将一直等待缓冲区直至其变为可用。不然若是producer生产速度过快耗尽了缓冲区，producer将抛出异常
• acks=all  很好理解，全部follower都响应了才认为消息提交成功，即"committed"
• retries = MAX 无限重试，直到你意识到出现了问题:)
• max.in.flight.requests.per.connection = 1 限制客户端在单个链接上可以发送的未响应请求的个数。设置此值是1表示kafka broker在响应请求以前client不能再向同一个broker发送请求。注意：设置此参数是为了不消息乱序
• 使用KafkaProducer.send(record, callback)而不是send(record)方法   自定义回调逻辑处理消息发送失败
• callback逻辑中最好显式关闭producer：close(0) 注意：设置此参数是为了不消息乱序
• unclean.leader.election.enable=false   关闭unclean leader选举，即不容许非ISR中的副本被选举为leader，以免数据丢失
• replication.factor >= 3   这个彻底是我的建议了，参考了Hadoop及业界通用的三备份原则
• min.insync.replicas > 1 消息至少要被写入到这么多副本才算成功，也是提高数据持久性的一个参数。与acks配合使用
• 保证replication.factor > min.insync.replicas  若是二者相等，当一个副本挂掉了分区也就无法正常工做了。一般设置replication.factor = min.insync.replicas + 1便可

2. Consumer端

　　consumer端丢失消息的情形比较简单：若是在消息处理完成前就提交了offset，那么就有可能形成数据的丢失。因为Kafka consumer默认是自动提交位移的，因此在后台提交位移前必定要保证消息被正常处理了，所以不建议采用很重的处理逻辑，若是处理耗时很长，则建议把逻辑放到另外一个线程中去作。为了不数据丢失，现给出两点建议：

• enable.auto.commit=false  关闭自动提交位移
• 在消息被完整处理以后再手动提交位移