如何保证kafka消息不丢失_max.in.flight.requests.per.connection java

如何保证kafka消息不丢失

我们从 producer、 consumer、 broker三个角度进行分析。

消息传输保障

一般而言，消息中间件的消息传输保障有 3 个层级 如下图所示：

Kafka的消息传输保障机制很直观。

针对生产者而言，生产者发送消息后：

若消息被成功提交到日志文件，因为多副本机制，那么这个消息就不会丢失；

若消息没有被成功提交到日志文件，如遭遇到了网络波动等问题，生产者无法判断消息是否已经提交成功，此时生产者可以进行重试，这个过程可能造成消息的重复写入。

在这里kafka提供的消息传输保障为at least once。

针对消费者而言，消费者处理消息和提交消费位移的顺序很大程度上决定了消费者提供哪种消息传输保障。

若消费者拉取完消息后，应用逻辑先处理消息后提交消费位移，那么在消息处理之后且位移提交之前消费者宕机了，等它重新上线后，会从上一次位移提交的位置拉取，造成重复消费，此时对应at least once。

若消费者拉取完消息后，应用逻辑先提交消费位移后处理消息，那么在位移提交之后且消息处理完成之前消费者宕机了，等它重新上线后，会从已经提交的位移处拉取，造成消息丢失，此时对应at most once。

Producer端

消息的发送

kafkaProducer的send()方法并非void类型，而是Future<RecordMetadata>类型，它有两个重载方法：

public Future<RecordMetadata> send(ProducerRecord<K,V> record)
public Future<RecordMetadata> send(ProducerRecord<K,V> record,Callback callback)
1
2

send()方法本身是异步的，send()方法返回的Future对象可以使调用方稍后获得发送的结果。

同步发送的发送方式可以利用返回的Future对象实现，代码如下：

try {
		producer.send(record).get();
    } catch (Exception e) {
		e.printStackTrace();
    }
1
2
3
4
5

上述代码在执行send()方法之后直接链式调用了get()方法来阻塞等待kafka的响应，一直到消息发送成功，或者发生异常。如果有异常出现就得捕获异常然后交给外层逻辑处理。

另一种同步发送代码如下：

try {
    Future<RecordMetadata> send = producer.send(record);
    RecordMetadata recordMetadata = send.get();
    System.out.println(recordMetadata.topic()+recordMetadata.partition()+recordMetadata.offset());
    } catch (Exception e) {
    e.printStackTrace();
    }
1
2
3
4
5
6
7

从上述代码可知，我们可以获得一个RecordMetadata对象，它里面包含了消息的一些元数据，比如主题、分区、偏移量等。如果不需要这些数据则第一种方法更省事。

异步发送代码如下：

try {
    producer.send(record, new Callback() {
        @Override
        public void onCompletion(RecordMetadata recordMetadata, Exception e) {
            if (e != null) {
                //可以进行消息重发、记录异常等
                e.printStackTrace();
            } else {
                System.out.println(recordMetadata.topic() + recordMetadata.partition() + recordMetadata.offset());
            }
        }
    });
} catch (Exception e) {
    e.printStackTrace();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

由上述代码知：异步发送一般是在send()方法中指定一个Callback的回调函数，kafka在返回响应时调用该函数来实现异步的发送确认。

tips：对于同一个分区而言，如果两个消息中record1优先于record2发送，则kafka可以保证对应的Callback1优先于Callback2调用，即回调函数的调用也可以保证分区有序。

---

Producer解决方案

KafkaProducer中一般会有两种类型的异常：可重试的异常和不可重试的异常。例如因为网络波动而导致的异常就是可重试的异常。

针对可重试的异常，可以配置retries参数，该参数意为在规定的重试次数内自行恢复就不会抛出异常。默认值为0(推荐3)，配置方式如下：

prop.put(ProducerConfig.RETRIES_CONFIG,3)
1

注意：该配置项会影响到一些性能。比如会增加客户端对于异常的反馈时延。

当将上述配置设置为大于0的值的时候（可能会造成消息重复），如何避免消息重复的情况？

kafka从0.11.0.0版本开始引入了幂等和事务两个特性，以此来实现EOS（exactly once mantics，精确一次处理语义），保证重新发送不会导致消息在日志出现重复。Broker 为Producer分配了一个ID（producer id，简称PID），并通过每条消息的序列号（sequence number）进行去重。事务用来保证多个分区写入操作的原子性。

其中启用幂等传递的方法配置：enable.idempotence = true。

启用事务支持的方法配置：设置属性 transcational.id = 自己定义。

另外有一个配置项需要注意：**max.in.flight.requests.per.connection**配置意为：限制每个连接（也就是客户端与 Node 之间的连接）最多缓存的请求数，默认为5。这个配置大于1的时候会出现错序现象（Kafka保证同一个分区内消息是有序的）：

• 如果第一批消息写入失败，而第二批消息写入成功，那么生产者会重新发送第一批的消息。如果成功发送，则将造成这两批消息的错序。
• 若需要保证消息的顺序，则建议将max.in.flight.requests.per.connection设置为1，而不是把acks设置为0，不过这样依然会影响吞吐量。

与此相关还有一个配置项，即retry.backoff.ms参数，该参数用来设定两次重试之间的时间间隔，防止生产者过早放弃重试。

这些参数配合使用可以让生产者更稳妥地进行重试。

KafkaProducer发送消息后会进行commit（将消息写入日志），由于副本机制，则可以保证消息不丢失，至于多少副本收到这条消息生产者才会认为这条消息成功写入，可以由acks这个参数来指定（重点，它涉及到了消息的可靠性与吞吐量之间的权衡）。

prop.put(ProducerConfig.ACKS_CONFIG,"-1");
1

ack有三种类型的值（字符串类型）：

• acks=1。默认值为1，生产者发送消息之后，只要分区的leader副本成功写入则就会收到来自服务器端的成功响应。

  • 如果消息无法写入leader副本（leader副本崩溃、正在重新选举），那么生产者就会获得一个错误的响应。为了避免消息的丢失，生产者可以选择重发消息。至此，可能造成消息的重复生产，即at least once。
  • 如果消息写入leader副本并成功返回响应给生产者，但是在其他follower副本拉取之前leader副本崩溃，那么此时这个消息还是丢失了，因为新选举的leader副本中并没有这条消息。
  • acks设置为1，是消息可靠性和吞吐量之间的这种方案。

• acks=0。生产者发送消息之后不需要等待任何服务器端的相应。

  • 如果在消息从发送到写入kafka的过程中出现异常，导致Kafka没有接收这条消息，那么生产者也无法得知，消息也就丢失了。
  • 这个设置可以使吞吐量达到最大。

• acks=-1。生产者发送消息之后，需要等待ISR中所有的副本都成功写入消息之后才能收到来自服务器端的成功响应，这个配置可以达到最强的可靠性。

  • 如果在成功写入leader副本之后，与ISR中所有副本同步之前leader副本宕机了，则生产者会受到异常一次告知此次发送失败。
  • 注意：这并不代表消息就一定可靠谱！因为ISR中可能只有leader副本，此时就退化成了acks=1的情况。

  那么什么时候会出现ISR中只有leader副本的情况呢？

  当leader副本的消息流入速度很快，而follower副本的同步速度很慢，在某一个临界点时所有的follower副本都被剔除出了ISR集合。这时候acks=-1就会演变为acks=1的情景，这样就增大了消息丢失的风险。

  怎么处理这种情况？

  kafka（Broker端）提供了一个参数来应对这种情况：min.insync.replicas参数（默认为1），可以配合acks=-1来使用。这个参数指定了ISR集合中最小的副本数。如果不满足则会抛出异常。

  一般来说需要配置副本数>min.insync.replicas。例如配置副本数为3，min.insync.replicas配置为2。需要注意的是：该参数在提升可用性的时候会从侧面影响到kafka的可用性。例如ISR中只有一个leader副本，不配置该参数还能用，配置了就不能写入了。

  与此相关还有一个配置，即unclean.leader.election.enable，默认为false。意为：是否可以当leader下线的时候从非ISR集合中选举新的leader。如果为true，则会导致消息的丢失，设置为false则会影响可用性。具体配置需要根据需求灵活选择。

  总结：

  • acks=-1，保证所有副本都成功写入消息之后才能收到服务器端的成功响应。
  • min.insync.replicas>1，保证不会出现ISR集合只有leader副本的情况。
  • retries>=3，增加重试次数以保证消息的不丢失（可能造成消息重复）
  • retry.backoff.ms根据场景设置，配合retries参数进行重试。
  • max.in.flight.requests.per.connection=1，保证分区内消息的顺序性。
  • enable.idempotence = true：启用幂等传递的方法配置，防止重试造成消息重复发送。

  ---

Broker端

Kafka Broker 集群接收到数据后会将数据进行持久化存储到磁盘，消息都是先写入到页缓存，然后由操作系统负责具体的刷盘任务或者使用fsync强制刷盘，如果此时 Broker 宕机 ，且选举了一个落后 leader副本很多的follower副本成为新的leader副本，那么落后的消息数据就会丢失。

如何解决？

同步刷盘（不建议）:

通过log.flush.interval.message、log.flush.interval.ms 等参数来控制。同步刷盘可以提高消息的可靠性，防止由于机器掉电等异常造成处于页缓存而没有及时写入磁盘的消息丢失。但是会严重影响性能。

通过多副本的机制来保障（推荐）。具体配置如下：

• unclean.leader.election.enable：设置为false，拒绝从ISR集合之外的副本中选举leader副本。这样的话就不用担心选举出来的副本消息落后原leader副本太多。
• replication.factor：设置大于等于3，这样当leader副本宕机之后才有follower副本选举成leader。
• min.insync.replicas：设置为大于1，指定ISR集合中最小的副本数，保证了ISR集合中不会只有leader副本的情况出现。注意：副本数>min.insync.replicas。

---

Consumer端

针对Consumer端，有一个配置需要特别注意：enable.auto.commit。该配置默认为true，即开启自动位移提交功能，此方式非常简便，但是会造成重复消费和消费丢失的问题。

这个默认的自动提交是定期提交，由客户端参数auto.commit.interval.ms控制，默认5s，到时会将拉渠道的每个分区中最大的消息位移进行提交。

我们可以将其设置为false来执行手动提交。

如果因为应用解析消息的异常，可能导致一部分消息一直无法消费，这时候可以将这类消息暂存在死信队列，以免影响整体的消费进度。

当我们使用手动提交位移的时候会遇到两种情况：

• 先处理消息，再提交位移（推荐）

  如果在处理消息的时候宕机了，由于没有成功提交位移，等Consumer重启后会从上次的offset重新拉取消息。造成重复消费的问题，需要业务保证幂等性。

• 先提交位移，再处理消息
  如果提交了位移，处理消息的时候宕机了，由于位移已经提交，当Consumer重启后会从已经提交的offset出开始消费，之前未处理完成的消息不会再被处理，对于这个Consumer而言消息已经丢失了。

在此处有一个原则：如果消息没有被成功消费，则不能提交所对应的消费位移。

手动提交可以细分为同步提交和异步提交，即Consumer的commitSync()和commitAsync()这两个方法。以下代码均先处理消息，后提交位移！

同步提交代码如下：

while (IS_RUNNING.get()){
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records) {
        //逻辑处理
    }
    consumer.commitSync();
}
1
2
3
4
5
6
7

consumer.commitSync()会根据poll()拉取的最新位移来进行提交，只要不发生不可恢复的错误，它就会阻塞消费者线程知道位移提交完成，对于不可恢复的异常，则需要将其捕获并做针对性处理。

除此之外还可以按照分区的粒度划分提交位移的界限，具体代码如下：

try {
    while (IS_RUNNING.get()) {
        ConsumerRecords<String, String> records = consumer.poll(100);
        for (TopicPartition partition : records.partitions()) {
            List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
            for (ConsumerRecord<String, String> record : partitionRecords) {
                //业务逻辑
            }
            long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
            consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset + 1)));
        }
    }
} catch (Exception e) {
    e.printStackTrace();
} finally {
    consumer.close();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

与commitSync()相反，异步提交的方式在执行的时候消费者线程不会被阻塞，可能在提交消费位移的结果还没返回之前就开始了新的拉取操作，具体代码如下：

while (IS_RUNNING.get()) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records) {
        System.out.println("topic:" + record.topic() + ",offset:" + record.offset() + ",value:" + record.value());
    }
    consumer.commitAsync(new OffsetCommitCallback() {
        @Override
        public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception e) {
            if (e != null){
                e.printStackTrace();
            }else {
                logger.error("fail to commit offsets {}",offsets);
            }
        }
    });
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

当位移提交完成后会回调OffsetCommitCallback的onComplete()方法。

如果异步提交的时候失败了怎么办？

若提交失败后选择重试，那么则会造成重复消费的问题！（一般来说不需要重试，因为位移提交失败的情况很少见，进行重试会增加代码逻辑的复杂度）

例如：某一次提交位移为a，但是提交失败了，然后下一次又异步提交了位移a+b，这次成功了。如果使用重试机制，若前一次的异步提交的消费位移在重试的时候成功了，那么此时的消费位移也将变成a，如果此时再发生异常，那么消费者恢复之后就会从a开始消费。

怎么解决？

可以设置一个递增的序号来维护异步提交的顺序。每次提交之后就增加序号相对应的值。在提交失败之后检查所提交的位移与序号的值的大小，如果前者较小，则说明有更大的位移已经提交过，则不用进行重试。

如果在异步提交的情况下，消费者异常退出，则很可能导致消息重复消费的情况，因为这种情况下无法及时提交位移；如果消费者正常退出或者发再均衡，那么可以退出或者再均衡之前使用同步提交的方式做好把关。

总结：

为保证消息不丢失，开启配置enable.auto.commit为false，代码逻辑使用先处理消息，再提交位移的方式，针对消息重复消费问题，需要业务保证幂等性。