【kafka专栏】SpringBoot下`@KafkaListener`消费监听属性详解

• • 一、`@KafkaListener`属性概述
  • 二、结合自定义配置是实现监听(常规用法)
  • 三、指定Topic分区及偏移量进行消费
  • 五、指定监听器工厂（实现消息过滤）
  • 六、properties配置
  • 七、注解方式获取消息头及消息体
  • 八、消息转发

KafkaListener有若干的配置属性，这些配置属性使用或者是结合使用，可以方便快捷的帮助我们实现kafka消费者数据监听的需求。这里的属性比较多，先大概了解一下，后续我会介绍。

public @interface KafkaListener {

   
   String id() default "";

   
   String containerFactory() default "";

   
   String[] topics() default {};

   
   String topicPattern() default "";

   
   TopicPartition[] topicPartitions() default {};

   
   String containerGroup() default "";

   
   String errorHandler() default "";

   
   String groupId() default "";

   
   boolean idIsGroup() default true;

   
   String clientIdPrefix() default "";

   
   String beanRef() default "__listener";

   
   String concurrency() default "";

   
   String autoStartup() default "";

   
   String[] properties() default {};

   
   boolean splitIterables() default true;

   
   String contentTypeConverter() default "";

}

通常我们会把消费者监听的主题，消费者组名称，消费者组中消费者数量等常用信息做成自定义配置（而不是在代码中写死），如下所示：

myconsumer:
    topic: topic-a,topic-b
    group-id: group-demo
    concurrency: 5

下面的消费者监听器监听了两个topic：topic-a,topic-b（使用SpEL表达式逗号分割为字符串数组），该消费者组命名为group-demo，包含5个消费者线程并行消费。

@KafkaListener(topics = "#{'${myconsumer.topic}'.split(',')}",
        groupId = "${myconsumer.group-id}",
        concurrency="${myconsumer.concurrency}")
public void readMsg(ConsumerRecord consumerRecord) {
     //监听到数据之后，进行处理操作
}

在某些特殊场景下，希望消费Topic主题中的某几个分区（不是全部分区消费）。或者是针对某个分区从指定的偏移量开始消费。

@KafkaListener(topicPartitions =
        { @TopicPartition(topic = "topic-a", partitions = { "0", "1" }),
          @TopicPartition(topic = "topic-b", partitions = {"0","4"},partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "300"))
        })
public void readMsg(ConsumerRecord record) {
    
}

上面例子监听topic-a的0,1分区（可能包含不只2个分区）；监听topic-b的第0和4分区 ，并且第1分区从offset为300的开始消费;

@Configuration
public class KafkaInitialConfiguration {
 
    // 监听器工厂
    @Autowired
    private ConsumerFactory consumerFactory;
 
    // 配置一个消息过滤策略
    @Bean
    public ConcurrentKafkaListenerContainerFactory myFilterContainerFactory() {
        ConcurrentKafkaListenerContainerFactory factory = newConcurrentKafkaListenerContainerFactory();
        factory.setConsumerFactory(consumerFactory);
        // 被过滤的消息将被丢弃
        factory.setAckDiscarded(true);
        //设置消息过滤策略
        factory.setRecordFilterStrategy(new RecordFilterStrategy() {
            @Override
            public boolean filter(ConsumerRecord consumerRecord) {
                //这里做逻辑判断
                //返回true的消息将会被丢弃
                return true;
            }
        });
        return factory;
    }
}

使用方法，myFilterContainerFactory是上文中bean方法的名称。

@KafkaListener(containerFactory ="myFilterContainerFactory")

除了上面提到的一些配置属性，实际上apache kafka consumer支持的原生配置属性，要比Spring 提供的配置属性多得多。所有的apache kafka原生配置属性都可以通过properties配置来传递。

@KafkaListener(properties = {"enable.auto.commit:false","max.poll.interval.ms:6000" })

• @Payload：获取的是消息的消息体，也就是发送的数据内容
• @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY)：获取发送消息的key
• @Header(KafkaHeaders.RECEIVED_PARTITION_ID)：获取当前消息是从哪个分区中监听到的
• @Header(KafkaHeaders.RECEIVED_TOPIC)：获取监听的TopicName
• @Header(KafkaHeaders.RECEIVED_TIMESTAMP)：获取到数据时间的时间戳

@KafkaListener(topics = "topic-a")
public void  readMsg(@Payload String data,
                         @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) Integer key,
                         @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
                         @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                         @Header(KafkaHeaders.RECEIVED_TIMESTAMP) long ts) {

}

Spring-Kafka只需要通过一个@SendTo注解即可以实现消息的转发，被注解方法的return值即转发的消息内容：

@Component
public class KafkaConsumer {
    // 消费监听
    @KafkaListener(topics = {"topic1"})
    @SendTo("topic2")
    public String listen1(String data) {
        System.out.println("业务A收到消息：" + data);
        return data + "(已处理)";
    }
 
    // 消费监听
    @KafkaListener(topics = {"topic2"})
    public void listen2(String data) {
        System.out.println("业务B收到消息：" + data);
    }
}

• 消息发往topic1，topic1处理完成的数据通过SendTo发往topic2
• topic2如果收到数据，打印业务B收到消息： + 你发送的消息数据