前言

Kafka 提供了数据高可靠的特性，
但是如果使用不当，
你可能无法享受到这一特性，
今天我们就来看看如何正确的使用Kafka 保证数据的不会丢失吧！

生产者的正确的消息发送方式

Kafka为生产者生产消息提供了一个 send(msg) 方法,
另有一个重载的方法send(msg, callback),

send(msg)
该方法可以将一条消息发送出去，
但是对发送出去的消息没有掌控能力，
无法得知其最后是不是到达了Kafka，
所以这是一种不可靠的发送方式，
但是也因为客户端只需要负责发送，
所以具有较好的性能。
```
Future<RecordMetadata> future = producer.send(record)
```
上面的示例代码也可以看到，send返回的是一个 Future,
也就是说其实你是可以 Future.get()获取返回值的，
但这种同步的方式，基本上可以说是不会用到。

send(msg, callback)
该方法可以将一条消息发送出去，
并且可以从callback回调中得到该条消息的发送结果，
并且callback是异步回调，
所以在兼具性能的情况下，
也对消息具有比较好的掌控。

 ProducerRecord<byte[],byte[]> record = new ProducerRecord<byte[],byte[]>("the-topic", key, value);
 producer.send(myRecord,
           new Callback() {
               public void onCompletion(RecordMetadata metadata, Exception e) {
                   if(e != null) {
                      e.printStackTrace();
                   } else {
                      System.out.println("The offset of the record we just sent is: " + metadata.offset());
                   }
               }
           });

综上，如果要使数据不丢失，
首先你就的使用 send(msg, callback)来发送消息，
绝大多数情况下，我也建议你这么做。

生产者的配置

当我们通过 send(msg, callback) 是不是就意味着消息一定不丢失了呢？
答案明显是：不是的
我们接着上面，
send(msg, callback)里面 callback返回的成功，
到底是不是真的确保消息万无一失了？
其实这个返回的成功也是可以在生产者配置的：

 Properties props = new Properties();
 props.put("bootstrap.servers", "localhost:9092");
//*******重点*****************
 props.put("acks", "all");
 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

 Producer<String, String> producer = new KafkaProducer<>(props);
 for (int i = 0; i < 100; i++)
     producer.send(new ProducerRecord<String, String>("my-topic", Integer.toString(i), Integer.toString(i)));
 producer.close();

这段代码是生产者发送消息的一个例子，
其中没使用callback主要是这里callback不是重点，
我们的重点是props.put("acks", "all");
这个acks配置属性就是我们callback成功的具体含义:

acks=0
acks = 0如果设置为零，那么生产者将完全不会管服务器是否收到消息。
该记录将立即添加到套接字缓冲区中并视为已发送。
并且重试配置不会生效（因为客户端通常不会知道任何故障）。
返回值的偏移量将始终等于 -1。
该方式具有最大的吞吐量，
一般建议直接配合 send(msg)使用。
acks=1
当leader接受到消息就会直接给客户端返回成功，
一般情况下这种模式都能很好的保证数据的不丢失，
只有在laeder接受到数据，
然后还没来得及同步到follower，
就挂掉了才会导致数据的丢失，
这种概率还是比较小的。
这也是默认的选择方式，
兼具较好的吞吐和较高的可靠性
acks=all 或者 acks=-1
当leader接受到消息，并同步到了一定数量的follower，
才向生产者发生成功的消息，
同步到的follower数量由 broker 端的 min.insync.replicas 决定
除非一些不可抗力因素，
这种方式基本可以确保数据的完全不丢失。

Broker 端的配置

其实到这里，生产者端基本已经做好了数据不丢失的大部分准备，
但是有些东西是要配合 Broker 端一起，
才能达到预期的不丢失数据的，
比如我们上面说到的

min.insync.replicas 配置
我们上面知道了，
当生产者 acks = -1 的时候，
写入的副本数就必须 >= min.insync.replicas 数，
当达不到这个要求的时候，
生产者端会收到一个either NotEnoughReplicas or NotEnoughReplicasAfterAppend的异常。
所以我们这个参数必须不能大于 replication.factor 副本数。
否则生产者将无法写入任何数据，
一般建议 replication.factor 数要大于 min.insync.replicas,
比如3个机器的集群，设置 replication.factor = 3，
那么设置 min.insync.replicas = 2 就可以了，
这样既保证了数据写入的时候有一个副本的冗余，
也能保证在一些情况下，
某台Broker宕机导致数据无法达到3个副本时，
依然可以正常写入数据。
unclean.leader.election.enable
这里 Broker 端还有一个重要的配置就是 unclean.leader.election.enable = false
这个配置代表着一些数据落后比较多的 follower，
是否能在leader宕机后被选举成新的 leader
如果你设置成 true，
很明显，如果这样的follower成为新leader，
就会造成最新的一部分数据丢失掉，

重试 retries

上面已经基本完成了不丢数据的方方面面了，
但是有些东西不是我们能控制的，
比如网络抖动等不可抗拒的因素，
这时候重试次数就很关键了，
配置合适的retries重试次数，
和合适的retry.backoff.ms重试间隔时间，
将为我们的数据发送提供更高的稳定性，
当然如果实在发送不成功，怎么办呢？
一般我们也可以把发送不成功的数据保存在一个日志文件，
如果数据很重要，那就发送警告信息，
人工干预一下。

Kafka ——如何保证消息不会丢失

Kafka ——如何保证消息不会丢失

前言

生产者的正确的消息发送方式

生产者的配置

Broker 端的配置

重试 retries