`@KafkaListener` 注解

@KafkaListener 注解用于将 bean 方法指定为监听器容器的监听器。该 bean 被包装在一个 MessagingMessageListenerAdapter 中，该适配器配置了各种功能，例如转换器，如果需要，可以将数据转换为与方法参数匹配的数据。

您可以使用 SpEL（通过 #{…}）或属性占位符（${…}）在注解上配置大多数属性。有关更多信息，请参见 Javadoc。

记录监听器

@KafkaListener 注解提供了一种简单的 POJO 监听器机制。以下示例展示了如何使用它

public class Listener {

    @KafkaListener(id = "foo", topics = "myTopic", clientIdPrefix = "myClientId")
    public void listen(String data) {
        ...
    }

}

这种机制需要在您的一个 @Configuration 类上使用 @EnableKafka 注解，以及一个监听器容器工厂，用于配置底层的 ConcurrentMessageListenerContainer。默认情况下，预期有一个名为 kafkaListenerContainerFactory 的 bean。以下示例展示了如何使用 ConcurrentMessageListenerContainer

@Configuration
@EnableKafka
public class KafkaConfig {

    @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
                        kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
                                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3);
        factory.getContainerProperties().setPollTimeout(3000);
        return factory;
    }

    @Bean
    public ConsumerFactory<Integer, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        ...
        return props;
    }
}

请注意，要设置容器属性，您必须使用工厂上的 getContainerProperties() 方法。它用作注入到容器中的实际属性的模板。

从版本 2.1.1 开始，您现在可以为注解创建的消费者设置 client.id 属性。clientIdPrefix 后缀为 -n，其中 n 是一个整数，表示使用并发时容器的编号。

从版本 2.2 开始，您现在可以使用注解本身的属性来覆盖容器工厂的 concurrency 和 autoStartup 属性。这些属性可以是简单值、属性占位符或 SpEL 表达式。以下示例展示了如何做到这一点

@KafkaListener(id = "myListener", topics = "myTopic",
        autoStartup = "${listen.auto.start:true}", concurrency = "${listen.concurrency:3}")
public void listen(String data) {
    ...
}

显式分区分配

您还可以使用显式主题和分区（以及可选的初始偏移量）来配置 POJO 监听器。以下示例展示了如何做到这一点

@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) {
    ...
}

您可以在 partitions 或 partitionOffsets 属性中指定每个分区，但不能同时指定两者。

与大多数注解属性一样，您可以使用 SpEL 表达式；有关如何生成大量分区的示例，请参阅手动分配所有分区。

从版本 2.5.5 开始，您可以将初始偏移量应用于所有分配的分区

@KafkaListener(id = "thing3", topicPartitions =
        { @TopicPartition(topic = "topic1", partitions = { "0", "1" },
             partitionOffsets = @PartitionOffset(partition = "*", initialOffset = "0"))
        })
public void listen(ConsumerRecord<?, ?> record) {
    ...
}

* 通配符表示 partitions 属性中的所有分区。每个 @TopicPartition 中只能有一个带有通配符的 @PartitionOffset。

此外，当监听器实现 ConsumerSeekAware 时，即使使用手动分配，也会调用 onPartitionsAssigned。这允许您在此时执行任何任意 seek 操作。

从版本 2.6.4 开始，您可以指定一个逗号分隔的分区列表或分区范围

@KafkaListener(id = "pp", autoStartup = "false",
        topicPartitions = @TopicPartition(topic = "topic1",
                partitions = "0-5, 7, 10-15"))
public void process(String in) {
    ...
}

范围是包含的；上面的示例将分配分区 0, 1, 2, 3, 4, 5, 7, 10, 11, 12, 13, 14, 15。

相同的技术可用于指定初始偏移量

@KafkaListener(id = "thing3", topicPartitions =
        { @TopicPartition(topic = "topic1",
             partitionOffsets = @PartitionOffset(partition = "0-5", initialOffset = "0"))
        })
public void listen(ConsumerRecord<?, ?> record) {
    ...
}

初始偏移量将应用于所有 6 个分区。

从 3.2 开始，@PartitionOffset 支持 SeekPosition.END、SeekPosition.BEGINNING、SeekPosition.TIMESTAMP，seekPosition 匹配 SeekPosition 枚举名称

@KafkaListener(id = "seekPositionTime", topicPartitions = {
        @TopicPartition(topic = TOPIC_SEEK_POSITION, partitionOffsets = {
                @PartitionOffset(partition = "0", initialOffset = "723916800000", seekPosition = "TIMESTAMP"),
                @PartitionOffset(partition = "1", initialOffset = "0", seekPosition = "BEGINNING"),
                @PartitionOffset(partition = "2", initialOffset = "0", seekPosition = "END")
        })
})
public void listen(ConsumerRecord<?, ?> record) {
    ...
}

如果 seekPosition 设置为 END 或 BEGINNING，则会忽略 initialOffset 和 relativeToCurrent。如果 seekPosition 设置为 TIMESTAMP，则 initialOffset 表示时间戳。

手动确认

当使用手动 AckMode 时，您也可以向监听器提供 Acknowledgment。以下示例还展示了如何使用不同的容器工厂。

@KafkaListener(id = "cat", topics = "myTopic",
          containerFactory = "kafkaManualAckListenerContainerFactory")
public void listen(String data, Acknowledgment ack) {
    ...
    ack.acknowledge();
}

消费者记录元数据

最后，有关记录的元数据可从消息头中获取。您可以使用以下头名称来检索消息的头

KafkaHeaders.OFFSET
KafkaHeaders.RECEIVED_KEY
KafkaHeaders.RECEIVED_TOPIC
KafkaHeaders.RECEIVED_PARTITION
KafkaHeaders.RECEIVED_TIMESTAMP
KafkaHeaders.TIMESTAMP_TYPE

从 2.5 版本开始，如果传入的记录具有 null 键，则 RECEIVED_KEY 不存在；以前，头会填充 null 值。此更改是为了使框架与 spring-messaging 约定保持一致，在该约定中，null 值的头不存在。

以下示例展示了如何使用头

@KafkaListener(id = "qux", topicPattern = "myTopic1")
public void listen(@Payload String foo,
        @Header(name = KafkaHeaders.RECEIVED_KEY, required = false) Integer key,
        @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
        @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
        @Header(KafkaHeaders.RECEIVED_TIMESTAMP) long ts
        ) {
    ...
}

参数注解 (@Payload、@Header) 必须在监听器方法的具体实现上指定；如果它们在接口上定义，则不会被检测到。

从 2.5 版本开始，您可以使用 ConsumerRecordMetadata 参数接收记录元数据，而不是使用离散头。

@KafkaListener(...)
public void listen(String str, ConsumerRecordMetadata meta) {
    ...
}

它包含 ConsumerRecord 中的所有数据，除了键和值。

批处理监听器

从 1.1 版本开始，您可以配置 @KafkaListener 方法以接收从消费者轮询接收到的整个批处理消费者记录。

非阻塞重试不支持批处理监听器。

要配置监听器容器工厂以创建批处理监听器，您可以设置 batchListener 属性。以下示例展示了如何操作

@Bean
public KafkaListenerContainerFactory<?> batchFactory() {
    ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setBatchListener(true);  // <<<<<<<<<<<<<<<<<<<<<<<<<
   return factory;
}

从 2.8 版本开始，您可以使用 @KafkaListener 注解上的 batch 属性覆盖工厂的 batchListener 属性。这与对容器错误处理程序的更改相结合，允许同一个工厂用于记录和批处理监听器。

从版本 2.9.6 开始，容器工厂为 recordMessageConverter 和 batchMessageConverter 属性提供了单独的设置器。以前，只有一个属性 messageConverter，它适用于记录和批处理监听器。

以下示例展示了如何接收有效负载列表

@KafkaListener(id = "list", topics = "myTopic", containerFactory = "batchFactory")
public void listen(List<String> list) {
    ...
}

主题、分区、偏移量等信息都可以在与有效负载平行的标头中找到。以下示例展示了如何使用标头

@KafkaListener(id = "list", topics = "myTopic", containerFactory = "batchFactory")
public void listen(List<String> list,
        @Header(KafkaHeaders.RECEIVED_KEY) List<Integer> keys,
        @Header(KafkaHeaders.RECEIVED_PARTITION) List<Integer> partitions,
        @Header(KafkaHeaders.RECEIVED_TOPIC) List<String> topics,
        @Header(KafkaHeaders.OFFSET) List<Long> offsets) {
    ...
}

或者，您可以接收一个 List 的 Message<?> 对象，每个对象都包含每个消息的偏移量和其他详细信息，但它必须是方法中唯一定义的参数（除了使用手动提交时的可选 Acknowledgment 和/或 Consumer<?, ?> 参数）。以下示例展示了如何做到这一点

@KafkaListener(id = "listMsg", topics = "myTopic", containerFactory = "batchFactory")
public void listen1(List<Message<?>> list) {
    ...
}

@KafkaListener(id = "listMsgAck", topics = "myTopic", containerFactory = "batchFactory")
public void listen2(List<Message<?>> list, Acknowledgment ack) {
    ...
}

@KafkaListener(id = "listMsgAckConsumer", topics = "myTopic", containerFactory = "batchFactory")
public void listen3(List<Message<?>> list, Acknowledgment ack, Consumer<?, ?> consumer) {
    ...
}

在这种情况下，不会对有效负载执行任何转换。

如果 BatchMessagingMessageConverter 配置了 RecordMessageConverter，您也可以在 Message 参数中添加泛型类型，并且有效负载会被转换。有关更多信息，请参见使用批处理监听器进行有效负载转换。

您也可以接收一个 ConsumerRecord<?, ?> 对象列表，但它必须是方法中唯一定义的参数（除了使用手动提交时的可选 Acknowledgment 和 Consumer<?, ?> 参数）。以下示例展示了如何做到这一点

@KafkaListener(id = "listCRs", topics = "myTopic", containerFactory = "batchFactory")
public void listen(List<ConsumerRecord<Integer, String>> list) {
    ...
}

@KafkaListener(id = "listCRsAck", topics = "myTopic", containerFactory = "batchFactory")
public void listen(List<ConsumerRecord<Integer, String>> list, Acknowledgment ack) {
    ...
}

从版本 2.2 开始，监听器可以接收 poll() 方法返回的完整 ConsumerRecords<?, ?> 对象，让监听器访问其他方法，例如 partitions()（返回列表中的 TopicPartition 实例）和 records(TopicPartition)（获取选择性记录）。同样，这必须是方法中唯一定义的参数（除了使用手动提交时的可选 Acknowledgment 和 Consumer<?, ?> 参数）。以下示例展示了如何做到这一点

@KafkaListener(id = "pollResults", topics = "myTopic", containerFactory = "batchFactory")
public void pollResults(ConsumerRecords<?, ?> records) {
    ...
}

如果容器工厂配置了 RecordFilterStrategy，则它会被 ConsumerRecords<?, ?> 监听器忽略，并发出 WARN 日志消息。只有在使用 <List<?>> 形式的监听器时，才能使用批处理监听器过滤记录。默认情况下，记录会逐个过滤；从版本 2.8 开始，您可以覆盖 filterBatch 以一次性过滤整个批次。

注解属性

从版本 2.0 开始，id 属性（如果存在）将用作 Kafka 消费者 group.id 属性，覆盖消费者工厂中配置的属性（如果存在）。您也可以显式设置 groupId 或将 idIsGroup 设置为 false 以恢复以前使用消费者工厂 group.id 的行为。

您可以在大多数注解属性中使用属性占位符或 SpEL 表达式，如下例所示

@KafkaListener(topics = "${some.property}")

@KafkaListener(topics = "#{someBean.someProperty}",
    groupId = "#{someBean.someProperty}.group")

从版本 2.1.2 开始，SpEL 表达式支持一个特殊标记：__listener。它是一个伪 bean 名称，代表当前包含此注解的 bean 实例。

请考虑以下示例

@Bean
public Listener listener1() {
    return new Listener("topic1");
}

@Bean
public Listener listener2() {
    return new Listener("topic2");
}

鉴于之前示例中的 Bean，我们可以使用以下方法

public class Listener {

    private final String topic;

    public Listener(String topic) {
        this.topic = topic;
    }

    @KafkaListener(topics = "#{__listener.topic}",
        groupId = "#{__listener.topic}.group")
    public void listen(...) {
        ...
    }

    public String getTopic() {
        return this.topic;
    }

}

如果在极少数情况下，您有一个名为 __listener 的实际 Bean，您可以使用 beanRef 属性更改表达式标记。以下示例展示了如何操作

@KafkaListener(beanRef = "__x", topics = "#{__x.topic}", groupId = "#{__x.topic}.group")

从 2.2.4 版本开始，您可以在注解上直接指定 Kafka 消费者属性，这些属性将覆盖消费者工厂中配置的任何同名属性。您**不能**以这种方式指定 group.id 和 client.id 属性；它们将被忽略；请使用 groupId 和 clientIdPrefix 注解属性来指定这些属性。

属性以单独的字符串形式指定，使用标准的 Java Properties 文件格式：foo:bar、foo=bar 或 foo bar，如下例所示

@KafkaListener(topics = "myTopic", groupId = "group", properties = {
    "max.poll.interval.ms:60000",
    ConsumerConfig.MAX_POLL_RECORDS_CONFIG + "=100"
})

以下是使用 RoutingKafkaTemplate 中示例的相应监听器示例。

@KafkaListener(id = "one", topics = "one")
public void listen1(String in) {
    System.out.println("1: " + in);
}

@KafkaListener(id = "two", topics = "two",
        properties = "value.deserializer:org.apache.kafka.common.serialization.ByteArrayDeserializer")
public void listen2(byte[] in) {
    System.out.println("2: " + new String(in));
}

@KafkaListener 注解