小编Mak*_*lov的帖子

Spring Kafka 批量消费者的非阻塞重试

我正在使用 spring-kafka 2.8.0,我正在尝试为批量 kafka Consumer实现非阻塞重试。这是我的配置和消费者:

@Configuration
public class KafkaConfig {
    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, GenericRecord>> 
            batchListenerFactory(ConsumerFactory<Object, Object> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, GenericRecord> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.setBatchListener(true);
        return factory;
    }
}
Run Code Online (Sandbox Code Playgroud)
@Component
public class MyConsumer {

    @KafkaListener(
            topics = "my-topic",
            containerFactory = "batchListenerFactory"
    )
    @RetryableTopic(
            backoff = @Backoff(delay = 1000, multiplier = 2.0),
            attempts = "4",
            topicSuffixingStrategy = SUFFIX_WITH_INDEX_VALUE,
            autoCreateTopics = "false"
    )
    public void consume(List<ConsumerRecord<String, GenericRecord>> messages) {
        // do some stuff
    }

}
Run Code Online (Sandbox Code Playgroud)

但在 Stutup …

java spring apache-kafka spring-boot spring-kafka

3
推荐指数
1
解决办法
1571
查看次数

标签 统计

apache-kafka ×1

java ×1

spring ×1

spring-boot ×1

spring-kafka ×1