Kafka 无消息丢失配置如何实现？
1.如何保证kafka消息不丢失？
Kafka 只对“已提交”的消息（committed message）做有限度的持久化保证。
1.1“已提交”的消息 是什么？
Kafka 的若干个 Broker 成功地接收到一条消息并写入到日志文件后，它们会告诉生产者程序这条消息已成功提交。
1.2有限度的持久化保证 是什么？
就是在kafka限定的保存时间内并且物理硬件正常，存在不丢失。
2.kafka“消息丢失”案例
2.1 生产者程序丢失数据
2.1.1 问题描述：“发射后不管”
当你Producer想kafka发送了消息，但发现kafka没有保存。
KafkaProducer 类是异步发送消息的，如果你调用的是prodicer.send(msg)这个API接口，他会马上返回，但是并不能主管认为消息提交发送到kafka broker上的topic中。
@Override
public Future<RecordMetadata> send(ProducerRecord<K, V> record) {
return send(record, null);
}
这种方式的会造成“发射后不管”。
2.1.2 问题描述：“数据过大”
当发出的msg超过kafka规定接受大小，kafka当然不能接受到信息。
2.2 解决方案 使用自定义kafka回调类Callback
Producer 永远要使用带有回调通知的发送 API，也就是说不要使producer.send(msg)，而要使producer.send(msg, callback)
@Override
public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
// intercept the record, which can be potentially modified; this method does not throw exceptions
ProducerRecord<K, V> interceptedRecord = this.interceptors.onSend(record);
return doSend(interceptedRecord, callback);
}
使用kafka 自带的回调类Callback 自定义信息
kafkaProducer.send(new ProducerRecord(KAFKA_PRODUCER_TOPIC, data), new Callback() {
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
if (metadata!=null){
System.out.println("成功信息");
}else {
throw new RuntimeException("入库异常");
}
}
});
可以自定义得到类似结果
2.3 消费者程序丢失数据
很简单：维持先消费消息（阅读），再更新位移（书签）的顺序
当然会有可能导致消息的重复处理，但是总比丢失强。类似于同一页书被读了很多遍，但这不属于消息丢失的情形。
3.最后总结
3.1 Producer生产者
3.1.1 （Producer） acks=all
acks 是 Producer 的一个参数，代表了你对“已交”消息的定义。如果设置成 all，则表明所有副本Broker 都要接收到消息，该消息才算是“已提交”。这是最高等级的“已提交”定义。
3.1.2 使用 producer.send(msg, callback)
自定义回调信息，使用kafka client下的Callback
3.3 （Producer）设置 retries 为一个较大的值
这里的 retries 同样是 Producer 的参数，对应前面提到的 Producer 自动重试。当出现网络的瞬时抖动时，消息发送可能会失败，此时配置了retries > 0 的 Producer 能够自动重试消息发送，避免消息丢失。
3.2 Broker 端
3.2.1 unclean.leader.election.enable = false
这是 Broker 端的参数，它控制的是哪些 Broker 有资格竞选分区的 Leader。如果一个 Broker 落后原先的 Leader 太多，那么它一旦成为新的 Leader，必然会造成消息的丢失。故一般都要将该参数设置成 false，即不允许这种情况的发生。
3.2.2 replication.factor >= 3
这也是 Broker 端的参数。其实这里想表述的是，最好将消息多保存几份，毕竟目前防止消息丢失的主要机制就是冗余。
3.2.3 min.insync.replicas > 1
这依然是 Broker 端参数，控制的是消息至少要被写入到多少个副本才算是“已提交”。设置成大于 1 可以提升消息持久性。在实际环境中千万不要使用默认值 1。
3.2.4 replication.factor > min.insync.replicas
如果两者相等，那么只要有一个副本挂机，整个分区就无法正常工作了。我们不仅要改善消息的持久性，防止数据丢失，还要在不降低可用性的基础上完成。推荐设置成 replication.factor = min.insync.replicas +1。
3.3 Consumer 确保消息消费完成再提交
Consumer 端有个参数 enable.auto.commit，最好把它设置成 false，并采用手动提交位移的方式。就像前面说的，这对于单 Consumer 多线程处理的场景而言是至关重要的。