@KafkaListener详解与使用

@KafkaListener详解与使用

   

目录

1 id 监听器的id

2 监听topic

3 errorHandler 异常处理

4 containerFactory 监听器工厂

5 clientIdPrefix 客户端前缀

6 concurrency并发数

---

从2.2.4版开始，您可以直接在注释上指定Kafka使用者属性，这些属性将覆盖在使用者工厂中配置的具有相同名称的所有属性。

    可以使用#{…​}或属性占位符（${…​}）在SpEL上配置注释上的大多数属性。
    比如:

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

   

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

表示消费者的个数，一般和topic分区设置成一样。

1 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

    1

③.会覆盖消费者工厂的消费组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")

   

2 监听topic

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

可以同时监听多个
topics = {"SHI_TOPIC3","SHI_TOPIC4"}
topicPattern 支持表达式

    @KafkaListener(id = "pullPatternMsg", topicPattern = "rx_.*_.*_thing.*", concurrency = "1")
    @DependsOn("KafkaDataRouteConsumer")
    public void pullPatternMsg(@Payload String data,
                               @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                               @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) ByteBuffer key,
                               Acknowledgment ack,//手动提交offset
                               @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
                               @Header(KafkaHeaders.OFFSET) long offSet,
                             
                               Consumer consumer//消费者
    )

topicPartitions 显式分区分配

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

@KafkaListener(id = "thing2", topicPartitions =
        { @TopicPartition(topic = "topic1", partitions = { "0", "1" }),
          @TopicPartition(topic = "topic2", partitions = "0",
          partitionOffsets = @PartitionOffset(partition = "1", 
          initialOffset = "100"))
        })
public void listen(ConsumerRecord record) {
    ...
}

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

3 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"

4 containerFactory 监听器工厂

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

 举例： 批量消费的工厂类

   

 
    @Bean
    public KafkaListenerContainerFactory> batchFactory() {
        ConcurrentKafkaListenerContainerFactory factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(kafkaConsumerFactory());
        //设置为批量消费，每个批次数量在Kafka配置参数中设置ConsumerConfig.MAX_POLL_RECORDS_ConFIG
        factory.setBatchListener(true);
        return factory;
    }

使用containerFactory = "batchFactory"

5 clientIdPrefix 客户端前缀

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

6 concurrency并发数

    会覆盖消费者工厂中的concurrency ,这里的并发数就是多线程消费。