Kafka消息消费异常会如何处理?

1.前言

当我们使用@KafkaListener注解声明一个消费者时,该消费者就会轮询去拉取对应分区消息记录,消费消息记录,正如你所知道的那样,正常场景下会执行ack操作,提交offsetkafka服务器。但是异常场景下会如何执行,不知你是否也了解?在了解之前,先一起来看下异常处理器,看完之后想必会有所收获。

2.异常处理器

2.1 创建异常处理器

在实例化org.springframework.kafka.listener.KafkaMessageListenerContainer.ListenerConsumer#ListenerConsumer的时候如果没有自定义异常处理器,会去创建SeekToCurrentErrorHandler作为默认异常处理器使用

typescript

复制代码

protected ErrorHandler determineErrorHandler(GenericErrorHandler<?> errHandler) {
    return errHandler != null ? (ErrorHandler) errHandler
              : this.transactionManager != null
              ? null : new SeekToCurrentErrorHandler();
}

2.2 声明补偿策略

在构建默认异常处理器SeekToCurrentErrorHandler时会指定对应的补偿策略

csharp

复制代码

/**
 * Construct an instance with the default recoverer which simply logs the record after
 * {@value SeekUtils#DEFAULT_MAX_FAILURES} (maxFailures) have occurred for a
 * topic/partition/offset, with the default back off (9 retries, no delay).
 * [@since](https://learnku.com/users/65735) 2.2
 */
public SeekToCurrentErrorHandler() {
    this(null, SeekUtils.DEFAULT_BACK_OFF);
}

从代码注释上我们可以了解到该补偿策略会进行9次无时间间隔重试

2.3 调用异常处理器

scss

复制代码

@Nullable
private RuntimeException doInvokeRecordListener(final ConsumerRecord<K, V> record, // NOSONAR
 Iterator<ConsumerRecord<K, V>> iterator) {
 Object sample = startMicrometerSample();

 try {
 // 1.消费消息
 invokeOnMessage(record);
 successTimer(sample);
 recordInterceptAfter(record, null);
 } catch (RuntimeException e) {
 try {
 // 2.执行异常处理器
 invokeErrorHandler(record, iterator, e);
 // 3.提交offset
 commitOffsetsIfNeeded(record);
 } catch (KafkaException ke) {

 }
 }
 return null;
}

这里可以看到当消息消费异常后,会调用默认异常处理器SeekToCurrentErrorHandler

2.4 执行异常处理器

csharp

复制代码

public static boolean doSeeks(List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer, Exception exception,
 boolean recoverable, RecoveryStrategy recovery, @Nullable MessageListenerContainer container,
 LogAccessor logger) {
​
 Map<TopicPartition, Long> partitions = new LinkedHashMap<>();
 AtomicBoolean first = new AtomicBoolean(true);
 AtomicBoolean skipped = new AtomicBoolean();
 records.forEach(record -> {
 if (recoverable && first.get()) {
 try {
 // 1.判断该消息是否可重试
 boolean test = recovery.recovered(record, exception, container, consumer);
 skipped.set(test);
 }
 catch (Exception ex) {
 }
 }
 if (!recoverable || !first.get() || !skipped.get()) {
 partitions.computeIfAbsent(new TopicPartition(record.topic(), record.partition()),
 offset -> record.offset());
 }
 first.set(false);
 });
 // 2.重置分区偏移量,以便可以重复拉取异常消息
 seekPartitions(consumer, partitions, logger);
 return skipped.get();
}

异常处理器执行过程:

判断当前消息是否可重试

如果当前消息可以重试,会将该消息对应offset存储在partitions中,紧接着通过seekPartitions方法来将当前分区offset重置为当前消息offset,以至在下一次拉取消息的时候,仍然可以拉取到该异常消息。

如果当前消息不可以重试,判断此次拉取的消息是否只有一条,如果是,不做处理;如果不是,则通过partitions.computeIfAbsent方法设置分区offset为异常消息下一条消息对应offset,以至在下一次拉取的时候可以拉取到异常消息后的其它消息。

2.5 提交偏移量

csharp

复制代码

@Nullable
private RuntimeException doInvokeRecordListener(final ConsumerRecord<K, V> record, // NOSONAR
 Iterator<ConsumerRecord<K, V>> iterator) {
​
 Object sample = startMicrometerSample();try {
 invokeOnMessage(record);
 }
 catch (RuntimeException e) {
 try {
 invokeErrorHandler(record, iterator, e);
 // 提交分区offset
 commitOffsetsIfNeeded(record);
 } catch (KafkaException ke) {
 }
 }
 return null;
}

当默认异常处理器重试达到最大次数9次后,会执行commitOffsetsIfNeeded方法,手动提交分区offset

2.6 总结

当消费消息异常,在没有声明异常处理器的前提下会选择使用默认异常处理器SeekToCurrentErrorHandler,默认异常处理器会对异常消息进行重试,在达到最大重试次数9次后,会手动提交异常消息offset,然后继续消费异常消息之后的其它消息。

至此想必对消息消费异常有了一个大致认识,如有疑问,欢迎留言讨论。

作者:黑白搬砖工
链接:juejin.cn/post/7160650750995988493
来源:稀土掘金
著作权归作者所有。商业转载请联系作者获得授权,非商业转载请注明出处。

本作品采用《CC 协议》,转载必须注明作者和本文链接
MissYou-Coding
讨论数量: 0
(= ̄ω ̄=)··· 暂无内容!

讨论应以学习和精进为目的。请勿发布不友善或者负能量的内容,与人为善,比聪明更重要!
Coding Peasant @ 互联网
文章
175
粉丝
8
喜欢
51
收藏
59
排名:663
访问:1.1 万
私信
所有博文
博客标签
社区赞助商