spring-kafka @KafkaListener详解与使用

2022年6月6日09:58:46

说明

从2.2.4版开始，您可以直接在注释上指定Kafka使用者属性，这些属性将覆盖在使用者工厂中配置的具有相同名称的所有属性。您不能通过这种方式指定group.id和client.id属性。他们将被忽略；
可以使用#{…}或属性占位符（${…}）在SpEL上配置注释上的大多数属性。
比如:

@KafkaListener(id="consumer-id",topics="SHI_TOPIC1",concurrency="${listen.concurrency:3}",
            clientIdPrefix="myClientId")

属性concurrency将会从容器中获取listen.concurrency的值,如果不存在就默认用3

@KafkaListener详解

id 监听器的id

①. 消费者线程命名规则

填写:

2020-11-19 14:24:15 c.d.b.k.KafkaListeners 120 [INFO] 线程:Thread[consumer-id5-1-C-1,5,main]-groupId:BASE-DEMO consumer-id5 消费

没有填写ID:

2020-11-19 10:41:26 c.d.b.k.KafkaListeners 137 [INFO] 线程:Thread[org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1,5,main] consumer-id7

②.在相同容器中的监听器ID不能重复

否则会报错

Caused by: java.lang.IllegalStateException: Another endpoint is already registered with id

③.会覆盖消费者工厂的消费组GroupId

假如配置文件属性配置了消费组kafka.consumer.group-id=BASE-DEMO
正常情况它是该容器中的默认消费组
但是如果设置了@KafkaListener(id = "consumer-id7", topics = {"SHI_TOPIC3"})
那么当前消费者的消费组就是consumer-id7 ;

当然如果你不想要他作为groupId的话可以设置属性idIsGroup = false;那么还是会使用默认的GroupId;

④. 如果配置了属性groupId,则其优先级最高

@KafkaListener(id="consumer-id5",idIsGroup=false,topics="SHI_TOPIC3",groupId="groupId-test")

例如上面代码中最终这个消费者的消费组GroupId是 “groupId-test”

该id属性（如果存在）将用作Kafka消费者group.id属性,并覆盖消费者工厂中的已配置属性（如果存在）您还可以groupId显式设置或将其设置idIsGroup为false，以恢复使用使用者工厂的先前行为group.id。

groupId 消费组名

指定该消费组的消费组名; 关于消费组名的配置可以看看上面的id 监听器的id

如何获取消费者 group.id

在监听器中调用KafkaUtils.getConsumerGroupId()可以获得当前的groupId; 可以在日志中打印出来; 可以知道是哪个客户端消费的;

topics 指定要监听哪些topic(与topicPattern、topicPartitions 三选一)

可以同时监听多个
topics = {"SHI_TOPIC3","SHI_TOPIC4"}

topicPattern 匹配Topic进行监听(与topics、topicPartitions 三选一)

topicPartitions 显式分区分配

可以为监听器配置明确的主题和分区（以及可选的初始偏移量）

@KafkaListener(id="thing2", topicPartitions={@TopicPartition(topic="topic1", partitions={"0","1"}),@TopicPartition(topic="topic2", partitions="0",
             partitionOffsets=@PartitionOffset(partition="1", initialOffset="100"))})publicvoidlisten(ConsumerRecord<?,?>record){...}

上面例子意思是监听topic1的0,1分区；监听topic2的第0分区,并且第1分区从offset为100的开始消费;

errorHandler 异常处理

实现KafkaListenerErrorHandler; 然后做一些异常处理;

@Component
public class KafkaDefaultListenerErrorHandler implements KafkaListenerErrorHandler {
    @Override
    public Object handleError(Message<?> message, ListenerExecutionFailedException exception) {
        return null;
    }

    @Override
    public Object handleError(Message<?> message, ListenerExecutionFailedException exception, Consumer<?, ?> consumer) {
    	//do someting
        return null;
    }
}

调用的时候填写beanName;例如errorHandler="kafkaDefaultListenerErrorHandler"

containerFactory 监听器工厂

指定生成监听器的工厂类;

例如我写一个批量消费的工厂类

/**
     * 监听器工厂 批量消费
     * @return
     */@BeanpublicKafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer,String>>batchFactory(){ConcurrentKafkaListenerContainerFactory<Integer,String> factory=newConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(kafkaConsumerFactory());//设置为批量消费，每个批次数量在Kafka配置参数中设置ConsumerConfig.MAX_POLL_RECORDS_CONFIG
        factory.setBatchListener(true);return factory;}

使用containerFactory = "batchFactory"

clientIdPrefix 客户端前缀

会覆盖消费者工厂的kafka.consumer.client-id属性; 最为前缀后面接-n n是数字

concurrency并发数

会覆盖消费者工厂中的concurrency ,这里的并发数就是多线程消费; 比如说单机情况下,你设置了3; 相当于就是启动了3个客户端来分配消费分区;分布式情况总线程数=concurrency*机器数量; 并不是设置越多越好,具体如何设置请看属性concurrency的作用及配置(RoundRobinAssignor 、RangeAssignor)

/**
     * 监听器工厂 
     * @return
     */@BeanpublicKafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer,String>>concurrencyFactory(){ConcurrentKafkaListenerContainerFactory<Integer,String> factory=newConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(kafkaConsumerFactory());
        factory.setConcurrency(6);return factory;}

@KafkaListener(id="consumer-id5",idIsGroup=false,topics="SHI_TOPIC3", containerFactory="concurrencyFactory",concurrency= "1)

虽然使用的工厂是concurrencyFactory(concurrency配置了6); 但是他最终生成的监听器数量是1;

properties 配置其他属性

kafka中的属性看org.apache.kafka.clients.consumer.ConsumerConfig ;
同名的都可以修改掉;

用法

@KafkaListener(id="consumer-id5",idIsGroup=false,topics="SHI_TOPIC3", containerFactory="concurrencyFactory",concurrency="1", clientIdPrefix="myClientId5",groupId="groupId-test",
            properties={"enable.auto.commit:false","max.poll.interval.ms:6000"},errorHandler="kafkaDefaultListenerErrorHandler")

@KafkaListener使用

KafkaListenerEndpointRegistry

@AutowiredprivateKafkaListenerEndpointRegistry registry;//.... 获取所有注册的监听器
        registry.getAllListenerContainers();

设置入参验证器

当您将Spring Boot与验证启动器一起使用时，将LocalValidatorFactoryBean自动配置：如下

@Configuration@EnableKafkapublicclassConfigimplementsKafkaListenerConfigurer{@AutowiredprivateLocalValidatorFactoryBean validator;...@OverridepublicvoidconfigureKafkaListeners(KafkaListenerEndpointRegistrar registrar){
      registrar.setValidator(this.validator);}}

使用

@KafkaListener(id="validated", topics="annotated35", errorHandler="validationErrorHandler",
      containerFactory="kafkaJsonListenerContainerFactory")publicvoidvalidatedListener(@Payload@ValidValidatedClass val){...}@BeanpublicKafkaListenerErrorHandlervalidationErrorHandler(){return(m, e)->{...};}

TODO…

说明