@KafkaListener注释用于将bean方法指定为侦听器容器的侦听器。bean被包装在一个配置了各种功能的MessagingMessageListenerAdapter中，例如在必要时转换数据以匹配方法参数的转换器。
您可以使用#{…}或属性占位符（${…}）使用SpEL配置注释上的大多数属性。有关更多信息，请参阅Javadoc。
Record Listeners
@KafkaListener注释为简单的POJO监听器提供了一种机制。以下示例显示了如何使用它：
public class Listener {
@KafkaListener(id = "foo", topics = "myTopic", clientIdPrefix = "myClientId")
public void listen(String data) {
...
}
}此机制需要在您的@Configuration类之一上添加@EnableKafka注释和侦听器容器工厂，用于配置底层ConcurrentMessageListenerContainer。默认情况下，需要一个名为kafkaListencerContainerFactory的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版本开始，您现在可以通过使用注释本身的属性来覆盖容器工厂的并发性和autoStartup属性。属性可以是简单的值、属性占位符或SpEL表达式。以下示例显示了如何执行此操作：
@KafkaListener(id = "myListener", topics = "myTopic",
autoStartup = "${listen.auto.start:true}", concurrency = "${listen.concurrency:3}")
public void listen(String data) {
...
}Explicit Partition Assignment
您还可以使用显式主题和分区（以及可选的初始偏移量）配置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时，即使使用手动分配，现在也会调用onPartionsAssigned。例如，这允许在当时进行任何任意的寻道操作。
从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。结束，寻找位置。开始，寻找位置。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表示时间戳。
Manual Acknowledgment
使用手动确认模式时，您还可以向听众提供确认。要激活手动确认模式，您需要将ContainerProperties中的确认模式设置为相应的手动模式。以下示例还显示了如何使用不同的容器工厂。此自定义容器工厂必须通过调用getContainerProperties（），然后对其调用setAckMode，将AckMode设置为手动类型。否则，Acknowledgment对象将为null。
@KafkaListener(id = "cat", topics = "myTopic",
containerFactory = "kafkaManualAckListenerContainerFactory")
public void listen(String data, Acknowledgment ack) {
...
ack.acknowledge();
}Consumer Record Metadata
最后，可以从消息头中获得有关记录的元数据。您可以使用以下标头名称来检索邮件的标头：
卡夫卡标头。抵消
卡夫卡标头。已接收_KEY
卡夫卡标头。RECEIVED_TOPIC
卡夫卡标头。已接收_分区
卡夫卡标头。已接收_时间戳
卡夫卡标头。时间戳类型
从版本2.5开始，如果传入记录具有空键，则RECEIVED_KEY不存在；以前，标头填充了null值。此更改是为了使框架与不存在空值标头的spring消息传递约定保持一致。
以下示例显示了如何使用标头：
@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中除键和值之外的所有数据。
Batch Listeners
从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) {
...
}或者，您可以接收消息列表<？>每个消息中都有每个偏移量和其他详细信息的对象，但它必须是方法上定义的唯一参数（使用手动提交时，除了可选的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版本开始，侦听器可以接收完整的ConsumerRecords<？，？>poll（）方法返回的对象，让监听器访问其他方法，如partitions（）（返回列表中的TopicPartition实例）和records（TopicPartition）（获取选择性记录）。同样，这必须是方法上的唯一参数（除了可选的Acknowledgment，当使用手动提交或Consumer<？，？>参数时）。以下示例显示了如何执行此操作：
@KafkaListener(id = "pollResults", topics = "myTopic", containerFactory = "batchFactory")
public void pollResults(ConsumerRecords<?, ?> records) {
...
}如果容器工厂配置了RecordFilterStrategy，则ConsumerRecords会忽略它？，？>侦听器，并发出警告日志消息。如果列表<？，则只能使用批处理侦听器筛选记录？>使用听众的形式。默认情况下，记录一次过滤一个；从2.8版本开始，您可以重写filterBatch，在一次调用中过滤整个批处理。
Annotation Properties
从2.0版本开始，id属性（如果存在）用作Kafka consumer 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属性文件格式的单个字符串：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));
}