@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 开始，你现在可以为该注解创建的 consumer 设置 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) {
...
}
主题分区分配
你可以通过三种方式为 @KafkaListener 配置主题。你必须以其中一种方式配置主题。
@KafkaListener(id = "myListener", topics = "myTopic")
public void listen(String data) {
...
}
@KafkaListener(id = "myListener", topicPattern = "my.*")
public void listen(String data) {
...
}
@KafkaListener(id = "myListener", topicPartitions = { @TopicPartition(topic = "myTopic", partitions = { "0", "1" })})
public void listen(String data) {
...
}
你可以直接按名称简单地配置主题。在这种情况下，你还可以配置多个主题，如 topics = {"myTopic1", myTopic2"}。
你也可以使用 topicPattern 配置主题，它支持基于正则表达式的主题订阅。
当你以这两种方式之一（主题或主题模式）配置主题时，Kafka 会根据 consumer 组自动分配分区。或者，你可以为 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。例如，这允许在那时执行任意的查找操作。
从版本 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。要激活手动 AckMode，你需要将 ContainerProperties 中的 ack-mode 设置为合适的手动模式。以下示例还展示了如何使用不同的容器工厂。这个自定义容器工厂必须通过调用 getContainerProperties() 然后在其上调用 setAckMode，将 AckMode 设置为手动类型。否则，Acknowledgment 对象将为 null。
@KafkaListener(id = "cat", topics = "myTopic",
containerFactory = "kafkaManualAckListenerContainerFactory")
public void listen(String data, Acknowledgment ack) {
...
ack.acknowledge();
}
从版本 4.1 开始，你可以使用 ackMode 属性直接在 @KafkaListener 注解上覆盖容器工厂的默认 AckMode：
@KafkaListener(id = "manual", topics = "myTopic", ackMode = "MANUAL")
public void listen(String data, Acknowledgment ack) {
...
ack.acknowledge();
}
ackMode 属性接受与 ContainerProperties.AckMode 枚举值对应的字符串值。这样就无需仅为不同的确认模式而创建单独的容器工厂。该属性也可以配置为 SpEL 表达式（#{…}）或属性占位符（${…}）。
Consumer 记录元数据
最后，关于记录的元数据可以从消息 header 中获得。你可以使用以下 header 名来检索消息的 header：
KafkaHeaders.OFFSETKafkaHeaders.RECEIVED_KEYKafkaHeaders.RECEIVED_TOPICKafkaHeaders.RECEIVED_PARTITIONKafkaHeaders.RECEIVED_TIMESTAMPKafkaHeaders.TIMESTAMP_TYPE
从版本 2.5 开始，如果入站记录的 key 为 null，则 RECEIVED_KEY 不存在；以前该 header 会被填充为 null 值。这一改动是为了使框架与 spring-messaging 的约定保持一致，即值为 null 的 header 不存在。
以下示例展示了如何使用这些 header：
@KafkaListener(id = "qux", topics = "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 参数中接收记录元数据，而不是使用离散的 header。
@KafkaListener(...)
public void listen(String str, ConsumerRecordMetadata meta) {
...
}
这包含了 ConsumerRecord 中除 key 和 value 之外的所有数据。
批监听器
从版本 1.1 开始，你可以将 @KafkaListener 方法配置为接收从 consumer poll 接收到的整批 consumer 记录。
批监听器不支持非阻塞重试。
要将监听器容器工厂配置为创建批监听器，你可以设置 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 属性分别提供了单独的 setter。以前，只有一个属性 messageConverter，它同时适用于记录监听器和批监听器。
以下示例展示了如何接收一个 payload 列表：
@KafkaListener(id = "list", topics = "myTopic", containerFactory = "batchFactory")
public void listen(List<String> list) {
...
}
主题、分区、偏移量等都可在与 payload 平行的 header 中获得。以下示例展示了如何使用这些 header：
@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) {
...
}
或者，你可以接收一个 Message<?> 对象的 List，其中每条消息都带有各自的偏移量和其他详情，但它必须是该方法上定义的唯一参数（除了手动提交时的可选 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) {
...
}
在这种情况下，不会对 payload 执行任何转换。
如果 BatchMessagingMessageConverter 配置了一个 RecordMessageConverter，你还可以为 Message 参数添加泛型类型，从而对 payload 进行转换。有关更多信息，请参阅批监听器的 payload 转换。
你也可以接收一个 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 consumer 的 group.id 属性，覆盖 consumer 工厂中已配置的该属性（如果存在）。你也可以显式设置 groupId，或将 idIsGroup 设置为 false 来恢复以前使用 consumer 工厂 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 consumer 属性，这些属性将覆盖 consumer 工厂中配置的同名属性。你不能以这种方式指定 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));
}