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

1.前言

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

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 协议》，转载必须注明作者和本文链接