kafka怎样包管消息不丢失？

既然利用消息中心件就肯定涉及到一个紧张标题，怎样包管消息不丢失？
利用Kafka也不破例。简朴来说，一条消息会履历如下三个位置的流转：生产者、服务端、斲丧者。那么怎样包管消息不丢失也要从三个角度来思量：生产者发送消息、服务端存储消息、斲丧者斲丧消息。

1.发送消息角度

1.1利用有callback的send方法发送消息要利用准确方法。

Kafka 是异步发送消息的，也就是说我们调用 send 方法后立刻拿到告终果，并不代表这条消息肯定是发送乐成的。假如利用的是producer.send(msg)方法，就是如许的结果。这时间你有大概会发现方法明显调用成，但是Broker里却并没有收到这条消息。你大概会说，那把异步改成同步不就好了。由于send调用后会拿到一个feature对象，直接调用 get方法便会同步壅闭等候结果。要么消息发送乐成，要么发送失败抛出非常，根据非常做出相应的处置处罚，好比回滚、赔偿。但是壅闭等候结果会严肃影响性能，一样寻常并不实用。

1. producer.send(record).get();

复制代码

那么应该怎样包管发送的消息肯定被broker收到又要包管服从呢？
必要利用有回调（callback）的send 方法：

1. producer.send(msg, callback)。

复制代码

比方：

1. kafkaProducer.send(new ProducerRecord<>("first", "atguigu" + i), new Callback() {
2.     @Override
3.     public void onCompletion(RecordMetadata metadata, Exception exception) {
4.         if (exception == null){
5.             System.out.println("主题："+metadata.topic() + " 分区："+ metadata.partition());
6.         }
7.     }
8. });

复制代码

它仍旧是接纳异步发送，以是消息发送的吞吐量并不受影响。接纳回调函数的方式来得到消息发送的终极结果，终极是失败照旧乐成，进步了消息发送的可靠性。
它能准确地告诉你消息是否真的提交乐成了。一旦出现消息提交失败的环境，你就可以有针对性地举行处置处罚。
举例来说，假如是由于那些瞬时错误，那么仅仅让 Producer 重试就可以了；假如是消息不合格造成的，那么可以调解消息格式后再次发送。总之，处置处罚发送失败的责任在 Producer 端而非 Broker 端。
异步发送，接纳了callback的方式来回调，进步了消息发送的吞吐量，也肯定程度的进步了消息的可靠性，根据回调的结果来判断是否必要做相应的回滚逻辑。
1.2设置 ack=all

producer提供了三种消息确认的模式，通过设置acks来实现。
acks=1时（默认）
体现数据发送到Kafka后，颠末leader乐成吸收消息的简直认，才算发送乐成，假如leader宕机了，就会丢失数据。
acks=0时
体现生产者将数据发送出去就不管了，不等候任何返回。这种环境下数据传输服从最高，但是数据可靠性最低，当 server挂掉的时间就会丢数据；
acks=-1/all时
体现生产者必要等候ISR中的全部follower都确认吸收到数据后才算发送完成，如许数据不会丢失，因此可靠性最高，性能最低。
将ack参数设置为 all，表明全部副本 Broker 都要吸收到消息，该消息才算是“已提交”，如许可靠性最高。
1.3设置发送重试 retries和max.in.flight.requests.per.connection

当发送失败时，好比网络抖动的缘故原由，客户端会举行重试，重试的次数由retries指定。此参数默认设置为0，即：快速失败模式当发送失败时由客户端来处置处罚后续是否要举行继承发送。
假如设置``retries大于 0 而没有设置max.in.flight.requests.per.connection=1`则意味着放弃发送消息的次序性。

1. retries = 3
2. max.in.flight.requests.per.connection = 1

复制代码

如许设置后，发送客户端会最大重试3次，直到broker返回ack。同时只有一个毗连向broker发送数据，包管了数据的次序性。
2.Broker 角度

2.1设置 replication.factor >= 3

该参数体现分区副本的个数。

1. replication.factor =3

复制代码

发起设置大于1，至少有一个副本，随着业务的紧张性提拔可以增长副本数量。假如 Leader 副本出现不可用，Follower 副本会被推举为新的 Leader 副本继承提供服务。冗余存储数据，做备份本身就是一种提拔消息可靠性的办法。
2.2 设置 unclean.leader.election.enable = false


该参数体现有哪些 Follower 可以有资格被推举为 Leader 。

1. unclean.leader.election.enable = false

复制代码

假如一个 Follower 的数据落伍 Leader 太多，那么一旦它被推举为新的 Leader， 数据就会丢失，因此我们要将其设置为false，防止此类环境发生。
2.3设置 min.insync.replicas > 1

该参数体现消息至少要被写入乐成到 ISR 多少个副本才算"已提交"，默认值是1，发起设置 min.insync.replicas > 1,如许才可以提拔消息长期性，包管数据不丢失。

1. min.insync.replicas = 2

复制代码

别的我们还必要确保一下 replication.factor > min.insync.replicas，假如相称，只要有一个副本不可以，整个 partition 将无法正常提供服务。
为了包管消息长期性的同时还要包管可用性，保举设置成：replication.factor =min.insync.replicas +1, 最大限度包管体系可用性。
3.斲丧消息角度

3.1 消息处置处罚完成后再提交

起首要明确，consumer斲丧消息的时间是用 斲丧者位移(Consumer Offset) 来标记斲丧位置的。

一个partition可以被多个斲丧者组的差别斲丧者斲丧，而且分别记载其位移数据，如下图。

每次吸收到消息并处置处罚完业务逻辑斲丧完成后，我们要明确的告诉客户端斲丧的位移。留意是先斲丧然后更新位移。我们还必要设置

• 设置文件设置接纳手动提交位移的方式；
  enable.auto.commit = false
  2.无论是正常或非常斲丧竣过后，同步偏移量
  cousumer.commitSync();
  

假如先更新位移，然后再异步斲丧会导致假如业务逻辑失败，但是位移已经更新乐成，相称于这条失败的消息以后无法处置处罚。
留意假如是体系预期外的非常，不要catch住，如许也会导致斲丧失败后无法重试。
3.2 审慎利用批量斲丧，保举单条拉取Kafka

ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
斲丧者都是接纳主动拉取的方式来从 broker 得到消息的，假如是批量斲丧，好比一次斲丧10条数据，必要在失败的消息处准确的返回消息位移。好比处置处罚第5条消息处置处罚失败，那就必要返回位移 5，下次拉取消息重新从这条开始斲丧。
假如是多线程处置处罚斲丧，那么位移就更欠利益置处罚了，好比来2个线程，每个线程处置处罚5条，A处置处罚乐成2条，B处置处罚3条，位移该怎样返回？
保举利用单条消息拉取的模式，如许非常处置处罚和位移处置处罚都最方便，也没有并发的标题，性能上的影响一样寻常都可以担当。
额外提一句，消息的处置处罚都要留意幂等性，无论是位移导致的重新斲丧，照旧在Kafka rebalance时间导致的重新斲丧，都必要我们留意这个标题。