Kafka消费失败重试处理

亦凉 2022-08-31 04:42 2484阅读 0赞

默认情况下，Spring-Kafka 达到配置的重试次数时，【每条消息的失败重试时间，由配置的时间隔决定】Consumer 如果依然消费失败 ，那么该消息就会进入到死信队列。

在实际开发中，我们可以通过实现自定义的 SeekToCurrentErrorHandler ，当 Consumer 消费消息异常的时候，进行拦截处理。例如设置重试次数，在达到最大重试次数仍然失败后保存消息到数据库后续做人工处理。

## SeekToCurrentErrorHandler ##

SeekToCurrentErrorHandler可以针对消息的单条消费失败的消费重试处理。在消息消费失败时，SeekToCurrentErrorHandler 会将 调用 Kafka Consumer 的 seek(TopicPartition partition, long offset) 方法，将 Consumer 对于该消息对应的 TopicPartition 分区的本地进度设置成该消息的位置。这样，Consumer 在下次从 Kafka Broker 拉取消息的时候，又能重新拉取到这条消费失败的消息，并且是第一条。同时，Spring-Kafka 使用 FailedRecordTracker 对每个 Topic 的每个 TopicPartition 消费失败次数进行计数，这样相当于对该 TopicPartition 的第一条消费失败的消息的消费失败次数进行计数。另外，在 FailedRecordTracker 中，会调用 BackOff 来进行计算，该消息的下一次重新消费的时间，通过 Thread\#sleep(...) 方法，实现重新消费的时间间隔。

但是，FailedRecordTracker 提供的计数是客户端级别的，重启 JVM 应用后，计数是会丢失的。所以，如果想要计数进行持久化，需要自己重新实现下 FailedRecordTracker 类，通过 ZooKeeper 存储计数。

public class SeekToCurrentErrorHandler extends FailedRecordProcessor implements ContainerAwareErrorHandler {
        private boolean ackAfterHandle;
    
    	//默认是失败后重试9次，没有BiConsumer接口的实现也就是没有后续处理逻辑
        public SeekToCurrentErrorHandler() {
            this((BiConsumer)null, SeekUtils.DEFAULT_BACK_OFF);//DEFAULT_BACK_OFF = new FixedBackOff(0L, 9L)
        }
    
        public SeekToCurrentErrorHandler(BackOff backOff) {
            this((BiConsumer)null, backOff);
        }
    
        public SeekToCurrentErrorHandler(BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer) {
            this(recoverer, SeekUtils.DEFAULT_BACK_OFF);//DEFAULT_BACK_OFF = new FixedBackOff(0L, 9L)
        }
    
    	public SeekToCurrentErrorHandler(@Nullable BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer, BackOff backOff) {
    		super(recoverer, backOff);
    		this.ackAfterHandle = true;
    	}
        public void handle(Exception thrownException, List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer, MessageListenerContainer container) {
            SeekUtils.seekOrRecover(thrownException, records, consumer, container, this.isCommitRecovered(), this.getSkipPredicate(records, thrownException), this.logger, this.getLogLevel());
        }
    }

## 自定义SeekToCurrentErrorHandler处理消费异常 ##

通过上面的源码可以看到默认无参构造器指定了BackOff为SeekUtils.DEFAULT\_BACK\_OFF,即默认是失败后重试9次。但是默认没有BiConsumer接口的实现也就是没有后续处理逻辑。我们可以在项目中自定义SeekToCurrentErrorHandler这个Bean,然后自定义BiConsumer接口的实现来实现重试最大次数扔失败后的后续续处理逻辑和失败时重试次数。

@Configuration
    public class KafkaConfiguration {
    	@Bean
    	@Primary
    	public SeekToCurrentErrorHandler seekToCurrentErrorHandler() {
    		SeekToCurrentErrorHandler errorHandler = new SeekToCurrentErrorHandler(new BiConsumer<ConsumerRecord<?, ?>, Exception>() {
    			@Override
    			public void accept(ConsumerRecord<?, ?> consumerRecord, Exception e) {
    				//重试4次仍然失败后会进去BiConsumer接口的accept方法，可以进行保存数据库等操作
    				log.info("消费失败4次,消息保存到数据库. record:{}",JSON.toJSONString(record));
    			}
    		}, new FixedBackOff(0L, 4L));//interval 为0标识立即重试，maxAttempts为4标识最多重试四次
    		return errorHandler;
    	}
    }