# Spring for Apache Kafka

# 1.前言

Spring for Apache Kafka 项目将核心 Spring 概念应用于基于 Kafka 的消息传递解决方案的开发。我们提供了一个“模板”作为发送消息的高级抽象。我们还为消息驱动的 POJO 提供支持。

# 2.最新更新?

# 2.1. 自从2.7之后2.8中的更新

本部分介绍了从 2.7 版本到 2.8 版本所做的更改。有关早期版本中的更改，请参见 [更新历史] 。

# 2.1.1.Kafka 客户端版本

此版本需要 3.0.0 kafka-clients

有关更多信息，请参见 一次语义学 和 KIP-447 (opens new window) 。

# 2.1.2.软件包更改

与类型映射相关的类和接口已从 …​support.converter 移动到 …​support.mapping 。

• AbstractJavaTypeMapper

• ClassMapper

• DefaultJackson2JavaTypeMapper

• Jackson2JavaTypeMapper

# 2.1.3.失效的手动提交

现在可以将侦听器容器配置为接受顺序错误的手动偏移提交（通常是异步的）。容器将推迟提交，直到确认丢失的偏移量。有关更多信息，请参见 手动提交偏移 。

# 2.1.4. @KafkaListener 变化

现在可以在方法本身上指定侦听器方法是否为批处理侦听器。这允许对记录和批处理侦听器使用相同的容器工厂。

有关更多信息，请参见 批处理侦听器 。

批处理侦听器现在可以处理转换异常。

有关更多信息，请参见 使用批处理错误处理程序的转换错误 。

RecordFilterStrategy 在与批处理侦听器一起使用时，现在可以在一个调用中过滤整个批处理。有关更多信息，请参见 批处理侦听器 末尾的注释。

# 2.1.5. KafkaTemplate 变化

给定主题、分区和偏移量，你现在可以接收一条记录。有关更多信息，请参见[使用 KafkaTemplate 接收]。

# 2.1.6. CommonErrorHandler 已添加

遗留的 GenericErrorHandler 及其用于记录批处理侦听器的子接口层次结构已被新的单一接口 CommonErrorHandler 所取代，其实现方式与 GenericErrorHandler 的大多数遗留实现方式相对应。有关更多信息，请参见 容器错误处理程序 。

# 2.1.7.监听器容器更改

默认情况下， interceptBeforeTx 容器属性现在是 true 。

authorizationExceptionRetryInterval 属性已重命名为 authExceptionRetryInterval ，并且现在除了以前的 AuthorizationException s 之外，还应用于 AuthenticationException s。这两个异常都被认为是致命的，除非设置了此属性，否则默认情况下容器将停止。

有关更多信息，请参见[使用 KafkaMessageListenerContainer ]和 侦听器容器属性 。

# 2.1.8.序列化器/反序列化器更改

现在提供了 DelegatingByTopicSerializer 和 DelegatingByTopicDeserializer 。有关更多信息，请参见 委派序列化器和反序列化器 。

# 2.1.9. DeadLetterPublishingRecover 变化

默认情况下，属性 stripPreviousExceptionHeaders 现在是 true 。

有关更多信息，请参见 管理死信记录头 。

# 2.1.10.可重排的主题更改

现在，你可以对可重试和不可重试的主题使用相同的工厂。有关更多信息，请参见 指定 ListenerContainerFactory 。

现在，全球范围内出现了一系列可控的致命异常，这些异常将使失败的记录直接流向 DLT。请参阅 异常分类器 以了解如何管理它。

使用可重排主题功能时引发的 KafkabackoffException 现在将在调试级别记录。如果需要更改日志级别以返回警告或将其设置为任何其他级别，请参见[[change-kboe-logging-level]]。

# 3.导言

参考文档的第一部分是对 Spring Apache Kafka 和底层概念以及一些代码片段的高级概述，这些代码片段可以帮助你尽快启动和运行。

# 3.1.快速游览

先决条件：你必须安装并运行 Apache Kafka。然后，你必须将 Apache Kafka（ spring-kafka ）的 Spring JAR 及其所有依赖项放在你的类路径上。最简单的方法是在构建工具中声明一个依赖项。

如果不使用 Spring boot，请在项目中将 spring-kafka jar 声明为依赖项。

Maven

<dependency>
  <groupId>org.springframework.kafka</groupId>
  <artifactId>spring-kafka</artifactId>
  <version>2.8.3</version>
</dependency>

Gradle

compile 'org.springframework.kafka:spring-kafka:2.8.3'

Maven

<dependency>
  <groupId>org.springframework.kafka</groupId>
  <artifactId>spring-kafka</artifactId>
</dependency>

Gradle

compile 'org.springframework.kafka:spring-kafka'

然而，最快的入门方法是使用 start.spring.io (opens new window) （或 Spring Tool Suits 和 IntelliJ Idea 中的向导）并创建一个项目，选择’ Spring for Apache Kafka’作为依赖项。

# 3.1.1.相容性

此快速浏览适用于以下版本：

• Apache Kafka Clients3.0.0

• Spring Framework5.3.x

• 最低 Java 版本：8

# 3.1.2.开始

最简单的入门方法是使用 start.spring.io (opens new window) （或 Spring Tool Suits 和 IntelliJ Idea 中的向导）并创建一个项目，选择’ Spring for Apache Kafka’作为依赖项。请参阅 Spring Boot documentation (opens new window) 以获取有关其对基础设施 bean 的自以为是的自动配置的更多信息。

这是一个最小的消费者应用程序。

# Spring 引导消费者应用程序

例 1.应用程序

Java

@SpringBootApplication
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    @Bean
    public NewTopic topic() {
        return TopicBuilder.name("topic1")
                .partitions(10)
                .replicas(1)
                .build();
    @KafkaListener(id = "myId", topics = "topic1")
    public void listen(String in) {
        System.out.println(in);

Kotlin

@SpringBootApplication
class Application {
    @Bean
    fun topic() = NewTopic("topic1", 10, 1)
    @KafkaListener(id = "myId", topics = ["topic1"])
    fun listen(value: String?) {
        println(value)
fun main(args: Array<String>) = runApplication<Application>(*args)

示例 2.application.properties

spring.kafka.consumer.auto-offset-reset=earliest

NewTopic Bean 导致在代理上创建主题；如果主题已经存在，则不需要该主题。

# Spring Boot Producer app

例 3.应用程序

Java

@SpringBootApplication
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    @Bean
    public NewTopic topic() {
        return TopicBuilder.name("topic1")
                .partitions(10)
                .replicas(1)
                .build();
    @Bean
    public ApplicationRunner runner(KafkaTemplate<String, String> template) {
        return args -> {
            template.send("topic1", "test");

Kotlin

@SpringBootApplication
class Application {
    @Bean
    fun topic() = NewTopic("topic1", 10, 1)
    @Bean
    fun runner(template: KafkaTemplate<String?, String?>) =
        ApplicationRunner { template.send("topic1", "test") }
    companion object {
        @JvmStatic
        fun main(args: Array<String>) = runApplication<Application>(*args)

# 带 Java 配置（no Spring boot）

下面是一个不使用 Spring 引导的应用程序的示例；它同时具有 Consumer 和 Producer 。

例 4.没有引导

Java

public class Sender {
	public static void main(String[] args) {
		AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(Config.class);
		context.getBean(Sender.class).send("test", 42);
	private final KafkaTemplate<Integer, String> template;
	public Sender(KafkaTemplate<Integer, String> template) {
		this.template = template;
	public void send(String toSend, int key) {
		this.template.send("topic1", key, toSend);
public class Listener {
    @KafkaListener(id = "listen1", topics = "topic1")
    public void listen1(String in) {
        System.out.println(in);
@Configuration
@EnableKafka
public class Config {
    @Bean
    ConcurrentKafkaListenerContainerFactory<Integer, String>
                        kafkaListenerContainerFactory(ConsumerFactory<Integer, String> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
                                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        return factory;
    @Bean
    public ConsumerFactory<Integer, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerProps());
    private Map<String, Object> consumerProps() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // ...
        return props;
    @Bean
    public Sender sender(KafkaTemplate<Integer, String> template) {
        return new Sender(template);
    @Bean
    public Listener listener() {
        return new Listener();
    @Bean
    public ProducerFactory<Integer, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(senderProps());
    private Map<String, Object> senderProps() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.LINGER_MS_CONFIG, 10);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        //...
        return props;
    @Bean
    public KafkaTemplate<Integer, String> kafkaTemplate(ProducerFactory<Integer, String> producerFactory) {
        return new KafkaTemplate<Integer, String>(producerFactory);

Kotlin

class Sender(private val template: KafkaTemplate<Int, String>) {
    fun send(toSend: String, key: Int) {
        template.send("topic1", key, toSend)
class Listener {
    @KafkaListener(id = "listen1", topics = ["topic1"])
    fun listen1(`in`: String) {
        println(`in`)
@Configuration
@EnableKafka
class Config {
    @Bean
    fun kafkaListenerContainerFactory(consumerFactory: ConsumerFactory<Int, String>) =
        ConcurrentKafkaListenerContainerFactory<Int, String>().also { it.consumerFactory = consumerFactory }
    @Bean
    fun consumerFactory() = DefaultKafkaConsumerFactory<Int, String>(consumerProps)
    val consumerProps = mapOf(
        ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG to "localhost:9092",
        ConsumerConfig.GROUP_ID_CONFIG to "group",
        ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG to IntegerDeserializer::class.java,
        ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG to StringDeserializer::class.java,
        ConsumerConfig.AUTO_OFFSET_RESET_CONFIG to "earliest"
    @Bean
    fun sender(template: KafkaTemplate<Int, String>) = Sender(template)
    @Bean
    fun listener() = Listener()
    @Bean
    fun producerFactory() = DefaultKafkaProducerFactory<Int, String>(senderProps)
    val senderProps = mapOf(
        ProducerConfig.BOOTSTRAP_SERVERS_CONFIG to "localhost:9092",
        ProducerConfig.LINGER_MS_CONFIG to 10,
        ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG to IntegerSerializer::class.java,
        ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG to StringSerializer::class.java
    @Bean
    fun kafkaTemplate(producerFactory: ProducerFactory<Int, String>) = KafkaTemplate(producerFactory)

正如你所看到的，在不使用 Spring boot 时，你必须定义几个基础设施 bean。

# 4.参考文献

参考文档的这一部分详细介绍了构成 Spring Apache Kafka 的各种组件。 主要章节 涵盖了用 Spring 开发 Kafka 应用程序的核心类。

# 4.1.用 Spring 表示 Apache Kafka

这一部分提供了对使用 Spring 表示 Apache Kafka 的各种关注的详细解释。欲了解一个简短但不太详细的介绍，请参见 Quick Tour 。

# 4.1.1.连接到 Kafka

• KafkaAdmin -见 配置主题

• ProducerFactory -见 发送消息

• ConsumerFactory -见 接收消息

从版本 2.5 开始，每个扩展 KafkaResourceFactory 。这允许在运行时通过将 Supplier<String> 添加到它们的配置中来更改引导程序服务器： setBootstrapServersSupplier(() → …​) 。将对所有新连接调用该命令，以获取服务器列表。消费者和生产者通常都是长寿的。要关闭现有的生产者，请在 DefaultKafkaProducerFactory 上调用 reset() 。要关闭现有的消费者，在 stop() （然后 start() ）上调用 KafkaListenerEndpointRegistry 和/或 stop() ，并在任何其他侦听器容器 bean 上调用 start() 。

为了方便起见，该框架还提供了一个 ABSwitchCluster ，它支持两组引导程序服务器；其中一组在任何时候都是活动的。通过调用 setBootstrapServersSupplier() ，配置 ABSwitchCluster 并将其添加到生产者和消费者工厂，以及 KafkaAdmin 。当你想要切换时，在生产者工厂上调用 primary() 或 secondary() 并调用 reset() 以建立新的连接；对于消费者， stop() 和 start() 所有侦听器容器。当使用 @KafkaListener s， stop() 和 start() 时， KafkaListenerEndpointRegistry Bean。

有关更多信息，请参见 Javadocs。

# 工厂听众

从版本 2.5 开始， DefaultKafkaProducerFactory 和 DefaultKafkaConsumerFactory 可以配置为 Listener ，以便在创建或关闭生产者或消费者时接收通知。

生产者工厂监听器

interface Listener<K, V> {
    default void producerAdded(String id, Producer<K, V> producer) {
    default void producerRemoved(String id, Producer<K, V> producer) {

消费者工厂监听器

interface Listener<K, V> {
    default void consumerAdded(String id, Consumer<K, V> consumer) {
    default void consumerRemoved(String id, Consumer<K, V> consumer) {

在每种情况下， id 都是通过将 client-id 属性（创建后从 metrics() 获得）附加到工厂 beanName 属性中来创建的，并由 . 分隔。

例如，这些侦听器可用于在创建新客户机时创建和绑定 Micrometer KafkaClientMetrics 实例（并在客户机关闭时关闭它）。

该框架提供了可以做到这一点的侦听器；参见 千分尺本机度量 。

# 4.1.2.配置主题

如果你在应用程序上下文中定义了 KafkaAdmin Bean，那么它可以自动向代理添加主题。为此，你可以将每个主题的 NewTopic``@Bean 添加到应用程序上下文中。版本 2.3 引入了一个新的类 TopicBuilder ，以使创建这样的 bean 更加方便。下面的示例展示了如何做到这一点：

Java

@Bean
public KafkaAdmin admin() {
    Map<String, Object> configs = new HashMap<>();
    configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    return new KafkaAdmin(configs);
@Bean
public NewTopic topic1() {
    return TopicBuilder.name("thing1")
            .partitions(10)
            .replicas(3)
            .compact()
            .build();
@Bean
public NewTopic topic2() {
    return TopicBuilder.name("thing2")
            .partitions(10)
            .replicas(3)
            .config(TopicConfig.COMPRESSION_TYPE_CONFIG, "zstd")
            .build();
@Bean
public NewTopic topic3() {
    return TopicBuilder.name("thing3")
            .assignReplicas(0, Arrays.asList(0, 1))
            .assignReplicas(1, Arrays.asList(1, 2))
            .assignReplicas(2, Arrays.asList(2, 0))
            .config(TopicConfig.COMPRESSION_TYPE_CONFIG, "zstd")
            .build();

Kotlin

@Bean
fun admin() = KafkaAdmin(mapOf(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG to "localhost:9092"))
@Bean
fun topic1() =
    TopicBuilder.name("thing1")
        .partitions(10)
        .replicas(3)
        .compact()
        .build()
@Bean
fun topic2() =
    TopicBuilder.name("thing2")
        .partitions(10)
        .replicas(3)
        .config(TopicConfig.COMPRESSION_TYPE_CONFIG, "zstd")
        .build()
@Bean
fun topic3() =
    TopicBuilder.name("thing3")
        .assignReplicas(0, Arrays.asList(0, 1))
        .assignReplicas(1, Arrays.asList(1, 2))
        .assignReplicas(2, Arrays.asList(2, 0))
        .config(TopicConfig.COMPRESSION_TYPE_CONFIG, "zstd")
        .build()

从版本 2.6 开始，你可以省略 .partitions() 和/或 replicas() ，并且代理默认值将应用于这些属性。代理版本必须至少是 2.4.0 才能支持此功能-参见 KIP-464 (opens new window) 。

Java

@Bean
public NewTopic topic4() {
    return TopicBuilder.name("defaultBoth")
            .build();
@Bean
public NewTopic topic5() {
    return TopicBuilder.name("defaultPart")
            .replicas(1)
            .build();
@Bean
public NewTopic topic6() {
    return TopicBuilder.name("defaultRepl")
            .partitions(3)
            .build();

Kotlin

@Bean
fun topic4() = TopicBuilder.name("defaultBoth").build()
@Bean
fun topic5() = TopicBuilder.name("defaultPart").replicas(1).build()
@Bean
fun topic6() = TopicBuilder.name("defaultRepl").partitions(3).build()

从版本 2.7 开始，你可以在单个 KafkaAdmin.NewTopics Bean 定义中声明多个 NewTopic s:

Java

@Bean
public KafkaAdmin.NewTopics topics456() {
    return new NewTopics(
            TopicBuilder.name("defaultBoth")
                .build(),
            TopicBuilder.name("defaultPart")
                .replicas(1)
                .build(),
            TopicBuilder.name("defaultRepl")
                .partitions(3)
                .build());

Kotlin

@Bean
fun topics456() = KafkaAdmin.NewTopics(
    TopicBuilder.name("defaultBoth")
        .build(),
    TopicBuilder.name("defaultPart")
        .replicas(1)
        .build(),
    TopicBuilder.name("defaultRepl")
        .partitions(3)
        .build()

默认情况下，如果代理不可用，将记录一条消息，但将继续加载上下文。你可以通过编程方式调用管理员的 initialize() 方法稍后再试。如果你希望此条件被认为是致命的，请将管理员的 fatalIfBrokerNotAvailable 属性设置为 true 。然后，上下文将无法初始化。

从版本 2.7 开始， KafkaAdmin 提供了在运行时创建和检查主题的方法。

• createOrModifyTopics

• describeTopics

对于更高级的功能，你可以直接使用 AdminClient 。下面的示例展示了如何做到这一点：

@Autowired
private KafkaAdmin admin;
    AdminClient client = AdminClient.create(admin.getConfigurationProperties());
    client.close();

# 4.1.3.发送消息

本节介绍如何发送消息。

# 使用 KafkaTemplate

本节介绍如何使用 KafkaTemplate 发送消息。

# 概述

KafkaTemplate 封装了一个生成器，并提供了将数据发送到 Kafka 主题的方便方法。下面的清单显示了 KafkaTemplate 中的相关方法：

ListenableFuture<SendResult<K, V>> sendDefault(V data);
ListenableFuture<SendResult<K, V>> sendDefault(K key, V data);
ListenableFuture<SendResult<K, V>> sendDefault(Integer partition, K key, V data);
ListenableFuture<SendResult<K, V>> sendDefault(Integer partition, Long timestamp, K key, V data);
ListenableFuture<SendResult<K, V>> send(String topic, V data);
ListenableFuture<SendResult<K, V>> send(String topic, K key, V data);
ListenableFuture<SendResult<K, V>> send(String topic, Integer partition, K key, V data);
ListenableFuture<SendResult<K, V>> send(String topic, Integer partition, Long timestamp, K key, V data);
ListenableFuture<SendResult<K, V>> send(ProducerRecord<K, V> record);
ListenableFuture<SendResult<K, V>> send(Message<?> message);
Map<MetricName, ? extends Metric> metrics();
List<PartitionInfo> partitionsFor(String topic);
<T> T execute(ProducerCallback<K, V, T> callback);
// Flush the producer.
void flush();
interface ProducerCallback<K, V, T> {
    T doInKafka(Producer<K, V> producer);

有关更多详细信息，请参见 Javadoc (opens new window) 。

sendDefault API 要求为模板提供了一个默认的主题。

API 将 timestamp 作为参数，并将此时间戳存储在记录中。如何存储用户提供的时间戳取决于在 Kafka 主题上配置的时间戳类型。如果主题被配置为使用 CREATE_TIME ，则记录用户指定的时间戳（如果未指定，则生成时间戳）。如果将主题配置为使用 LOG_APPEND_TIME ，则忽略用户指定的时间戳，而代理添加本地代理时间。

metrics 和 partitionsFor 方法委托给底层[ Producer ]（https://kafka. Apache.org/20/javadoc/org/ Apache/kafka/clients/producer/producer.html）上相同的方法。 execute 方法提供了对底层[ Producer ]（https://kafka. Apache.org/20/javadoc/org/ Apache/kafka/clients/producer/producer.html）的直接访问。

要使用模板，你可以配置一个生产者工厂，并在模板的构造函数中提供它。下面的示例展示了如何做到这一点：

@Bean
public ProducerFactory<Integer, String> producerFactory() {
    return new DefaultKafkaProducerFactory<>(producerConfigs());
@Bean
public Map<String, Object> producerConfigs() {
    Map<String, Object> props = new HashMap<>();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    // See https://kafka.apache.org/documentation/#producerconfigs for more properties
    return props;
@Bean
public KafkaTemplate<Integer, String> kafkaTemplate() {
    return new KafkaTemplate<Integer, String>(producerFactory());

从版本 2.5 开始，你现在可以覆盖工厂的 ProducerConfig 属性，以创建具有来自同一工厂的不同生产者配置的模板。

@Bean
public KafkaTemplate<String, String> stringTemplate(ProducerFactory<String, String> pf) {
    return new KafkaTemplate<>(pf);
@Bean
public KafkaTemplate<String, byte[]> bytesTemplate(ProducerFactory<String, byte[]> pf) {
    return new KafkaTemplate<>(pf,
            Collections.singletonMap(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class));

注意，类型 ProducerFactory<?, ?> 的 Bean（例如由 Spring 引导自动配置的类型）可以用不同的窄泛型类型引用。

你还可以使用标准的 <bean/> 定义来配置模板。

然后，要使用模板，你可以调用它的一个方法。

当使用带有 Message<?> 参数的方法时，主题、分区和键信息将在包含以下项的消息头中提供：

• KafkaHeaders.TOPIC

• KafkaHeaders.PARTITION_ID

• KafkaHeaders.MESSAGE_KEY

• KafkaHeaders.TIMESTAMP

消息有效载荷就是数据。

可选地，你可以将 KafkaTemplate 配置为 ProducerListener ，以获得带有发送结果（成功或失败）的异步回调，而不是等待 Future 完成。下面的清单显示了 ProducerListener 接口的定义：

public interface ProducerListener<K, V> {
    void onSuccess(ProducerRecord<K, V> producerRecord, RecordMetadata recordMetadata);
    void onError(ProducerRecord<K, V> producerRecord, RecordMetadata recordMetadata,
            Exception exception);

默认情况下，模板配置为 LoggingProducerListener ，它会记录错误，并且在发送成功时不会执行任何操作。

为了方便起见，在你只想实现其中一个方法的情况下，提供了默认的方法实现。

注意，send 方法返回 ListenableFuture<SendResult> 。你可以向侦听器注册回调，以异步地接收发送的结果。下面的示例展示了如何做到这一点：

ListenableFuture<SendResult<Integer, String>> future = template.send("myTopic", "something");
future.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() {
    @Override
    public void onSuccess(SendResult<Integer, String> result) {
    @Override
    public void onFailure(Throwable ex) {

SendResult 有两个性质，a ProducerRecord 和 RecordMetadata 。有关这些对象的信息，请参见 Kafka API 文档。

Throwable 中的 onFailure 可以强制转换为 KafkaProducerException ；其 failedProducerRecord 属性包含失败的记录。

从版本 2.5 开始，你可以使用 KafkaSendCallback 而不是 ListenableFutureCallback ，从而更容易地提取失败的 ProducerRecord ，从而避免了强制转换 Throwable 的需要：

ListenableFuture<SendResult<Integer, String>> future = template.send("topic", 1, "thing");
future.addCallback(new KafkaSendCallback<Integer, String>() {
    @Override
    public void onSuccess(SendResult<Integer, String> result) {
    @Override
    public void onFailure(KafkaProducerException ex) {
        ProducerRecord<Integer, String> failed = ex.getFailedProducerRecord();

你也可以使用一对 lambdas:

ListenableFuture<SendResult<Integer, String>> future = template.send("topic", 1, "thing");
future.addCallback(result -> {
    }, (KafkaFailureCallback<Integer, String>) ex -> {
            ProducerRecord<Integer, String> failed = ex.getFailedProducerRecord();

如果你希望阻止发送线程以等待结果，则可以调用 Future 的 get() 方法；建议使用带有超时的方法。你可能希望在等待之前调用 flush() ，或者，为了方便起见，模板具有一个带有 autoFlush 参数的构造函数，该构造函数将在每次发送时使模板 flush() 。只有当你设置了 linger.ms producer 属性并希望立即发送部分批处理时，才需要刷新。

# 示例

本节展示了向 Kafka 发送消息的示例：

例 5.非阻塞（异步）

public void sendToKafka(final MyOutputData data) {
    final ProducerRecord<String, String> record = createRecord(data);
    ListenableFuture<SendResult<Integer, String>> future = template.send(record);
    future.addCallback(new KafkaSendCallback<Integer, String>() {
        @Override
        public void onSuccess(SendResult<Integer, String> result) {
            handleSuccess(data);
        @Override
        public void onFailure(KafkaProducerException ex) {
            handleFailure(data, record, ex);

阻塞（同步）

public void sendToKafka(final MyOutputData data) {
    final ProducerRecord<String, String> record = createRecord(data);
    try {
        template.send(record).get(10, TimeUnit.SECONDS);
        handleSuccess(data);
    catch (ExecutionException e) {
        handleFailure(data, record, e.getCause());
    catch (TimeoutException | InterruptedException e) {
        handleFailure(data, record, e);

注意， ExecutionException 的原因是 KafkaProducerException 具有 failedProducerRecord 属性。

# 使用 RoutingKafkaTemplate

从版本 2.5 开始，你可以使用 RoutingKafkaTemplate 在运行时基于目标 topic 名称选择生产者。

该模板需要一个 java.util.regex.Pattern 到 ProducerFactory<Object, Object> 实例的映射。这个映射应该是有序的（例如，a LinkedHashMap ），因为它是按顺序遍历的；你应该在开始时添加更具体的模式。

Spring 以下简单的引导应用程序提供了一个示例，说明如何使用相同的模板发送到不同的主题，每个主题使用不同的值序列化器。

@SpringBootApplication
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    @Bean
    public RoutingKafkaTemplate routingTemplate(GenericApplicationContext context,
            ProducerFactory<Object, Object> pf) {
        // Clone the PF with a different Serializer, register with Spring for shutdown
        Map<String, Object> configs = new HashMap<>(pf.getConfigurationProperties());
        configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
        DefaultKafkaProducerFactory<Object, Object> bytesPF = new DefaultKafkaProducerFactory<>(configs);
        context.registerBean(DefaultKafkaProducerFactory.class, "bytesPF", bytesPF);
        Map<Pattern, ProducerFactory<Object, Object>> map = new LinkedHashMap<>();
        map.put(Pattern.compile("two"), bytesPF);
        map.put(Pattern.compile(".+"), pf); // Default PF with StringSerializer
        return new RoutingKafkaTemplate(map);
    @Bean
    public ApplicationRunner runner(RoutingKafkaTemplate routingTemplate) {
        return args -> {
            routingTemplate.send("one", "thing1");
            routingTemplate.send("two", "thing2".getBytes());

该示例的相应 @KafkaListener s 如 注释属性 所示。

对于另一种实现类似结果的技术，但具有向相同主题发送不同类型的附加功能，请参见 委派序列化器和反序列化器 。

# 使用 DefaultKafkaProducerFactory

如[使用 KafkaTemplate ]（#kafka-template）中所示，使用 ProducerFactory 创建生产者。

当不使用 交易 时，默认情况下， DefaultKafkaProducerFactory 将创建一个由所有客户机使用的单例生成器，如 KafkaProducer Javadocs 中所建议的那样。但是，如果在模板上调用 flush() ，这可能会导致使用相同生成器的其他线程的延迟。从版本 2.3 开始， DefaultKafkaProducerFactory 有一个新的属性 producerPerThread 。当设置为 true 时，工厂将为每个线程创建（并缓存）一个单独的生产者，以避免此问题。

另请参见[ KafkaTemplate 事务性和非事务性发布]。

当创建 DefaultKafkaProducerFactory 时，可以通过调用只接收属性映射的构造函数（参见[using KafkaTemplate ]（#kafka-template）中的示例），从配置中获取键和/或值 Serializer 类，或者 Serializer 实例可以被传递到 DefaultKafkaProducerFactory 构造函数（在这种情况下，所有 Producer 的实例共享相同的实例）。或者，你可以提供 Supplier<Serializer> s（从版本 2.3 开始），它将用于为每个 Producer 获取单独的 Serializer 实例：

@Bean
public ProducerFactory<Integer, CustomValue> producerFactory() {
    return new DefaultKafkaProducerFactory<>(producerConfigs(), null, () -> new CustomValueSerializer());
@Bean
public KafkaTemplate<Integer, CustomValue> kafkaTemplate() {
    return new KafkaTemplate<Integer, CustomValue>(producerFactory());

从版本 2.5.10 开始，你现在可以在工厂创建后更新生产者属性。这可能是有用的，例如，如果你必须在凭据更改后更新 SSL 密钥/信任存储位置。这些更改将不会影响现有的生产者实例；调用 reset() 来关闭任何现有的生产者，以便使用新的属性创建新的生产者。注意：不能将事务性生产工厂更改为非事务性生产工厂，反之亦然。

现在提供了两种新的方法：

void updateConfigs(Map<String, Object> updates);
void removeConfig(String configKey);

从版本 2.8 开始，如果你将序列化器作为对象（在构造函数中或通过 setter）提供，则工厂将调用 configure() 方法来使用配置属性对它们进行配置。

# 使用 ReplyingKafkaTemplate

版本 2.1.3 引入了 KafkaTemplate 的子类来提供请求/回复语义。该类名为 ReplyingKafkaTemplate ，并具有两个附加方法；以下显示了方法签名：

RequestReplyFuture<K, V, R> sendAndReceive(ProducerRecord<K, V> record);
RequestReplyFuture<K, V, R> sendAndReceive(ProducerRecord<K, V> record,
    Duration replyTimeout);

（另请参见[request/reply with Message<?> s]（#exchange-messages））。

结果是一个 ListenableFuture ，该结果是异步填充的（或者是一个异常，用于超时）。结果还具有 sendFuture 属性，这是调用 KafkaTemplate.send() 的结果。你可以使用这个 future 来确定发送操作的结果。

如果使用第一个方法，或者 replyTimeout 参数是 null ，则使用模板的 defaultReplyTimeout 属性（默认情况下为 5 秒）。

Spring 以下引导应用程序显示了如何使用该功能的示例：

@SpringBootApplication
public class KRequestingApplication {
    public static void main(String[] args) {
        SpringApplication.run(KRequestingApplication.class, args).close();
    @Bean
    public ApplicationRunner runner(ReplyingKafkaTemplate<String, String, String> template) {
        return args -> {
            ProducerRecord<String, String> record = new ProducerRecord<>("kRequests", "foo");
            RequestReplyFuture<String, String, String> replyFuture = template.sendAndReceive(record);
            SendResult<String, String> sendResult = replyFuture.getSendFuture().get(10, TimeUnit.SECONDS);
            System.out.println("Sent ok: " + sendResult.getRecordMetadata());
            ConsumerRecord<String, String> consumerRecord = replyFuture.get(10, TimeUnit.SECONDS);
            System.out.println("Return value: " + consumerRecord.value());
    @Bean
    public ReplyingKafkaTemplate<String, String, String> replyingTemplate(
            ProducerFactory<String, String> pf,
            ConcurrentMessageListenerContainer<String, String> repliesContainer) {
        return new ReplyingKafkaTemplate<>(pf, repliesContainer);
    @Bean
    public ConcurrentMessageListenerContainer<String, String> repliesContainer(
            ConcurrentKafkaListenerContainerFactory<String, String> containerFactory) {
        ConcurrentMessageListenerContainer<String, String> repliesContainer =
                containerFactory.createContainer("kReplies");
        repliesContainer.getContainerProperties().setGroupId("repliesGroup");
        repliesContainer.setAutoStartup(false);
        return repliesContainer;
    @Bean
    public NewTopic kRequests() {
        return TopicBuilder.name("kRequests")
            .partitions(10)
            .replicas(2)
            .build();
    @Bean
    public NewTopic kReplies() {
        return TopicBuilder.name("kReplies")
            .partitions(10)
            .replicas(2)
            .build();

请注意，我们可以使用 Boot 的自动配置容器工厂来创建回复容器。

如果正在使用一个非平凡的反序列化器进行回复，请考虑使用一个[ ErrorHandlingDeserializer ]（#error-handling-deSerializer）将其委托给你配置的反序列化器。当这样配置时， RequestReplyFuture 将在特殊情况下完成，并且你可以捕获 ExecutionException ，而 DeserializationException 在其 cause 属性中。

从版本 2.6.7 开始，除了检测 DeserializationException s 之外，如果提供的话，模板将调用 replyErrorChecker 函数。如果它返回一个异常，则将来将异常完成。

下面是一个例子：

template.setReplyErrorChecker(record -> {
    Header error = record.headers().lastHeader("serverSentAnError");
    if (error != null) {
        return new MyException(new String(error.value()));
    else {
        return null;
RequestReplyFuture<Integer, String, String> future = template.sendAndReceive(record);
try {
    future.getSendFuture().get(10, TimeUnit.SECONDS); // send ok
    ConsumerRecord<Integer, String> consumerRecord = future.get(10, TimeUnit.SECONDS);
catch (InterruptedException e) {
catch (ExecutionException e) {
    if (e.getCause instanceof MyException) {
catch (TimeoutException e) {

模板设置一个头（默认情况下名为 KafkaHeaders.CORRELATION_ID ），必须由服务器端回显。

在这种情况下，以下 @KafkaListener 应用程序响应：

@SpringBootApplication
public class KReplyingApplication {
    public static void main(String[] args) {
        SpringApplication.run(KReplyingApplication.class, args);
    @KafkaListener(id="server", topics = "kRequests")
    @SendTo // use default replyTo expression
    public String listen(String in) {
        System.out.println("Server received: " + in);
        return in.toUpperCase();
    @Bean
    public NewTopic kRequests() {
        return TopicBuilder.name("kRequests")
            .partitions(10)
            .replicas(2)
            .build();
    @Bean // not required if Jackson is on the classpath
    public MessagingMessageConverter simpleMapperConverter() {
        MessagingMessageConverter messagingMessageConverter = new MessagingMessageConverter();
        messagingMessageConverter.setHeaderMapper(new SimpleKafkaHeaderMapper());
        return messagingMessageConverter;

@KafkaListener 基础结构与相关 ID 相呼应，并确定应答主题。

有关发送回复的更多信息，请参见[使用 @SendTo 转发侦听器结果]。该模板使用默认的头 KafKaHeaders.REPLY_TOPIC 来指示回复所针对的主题。

从版本 2.2 开始，模板将尝试从配置的应答容器中检测应答主题或分区。如果容器被配置为侦听单个主题或单个 TopicPartitionOffset ，则它将用于设置答复头。如果容器是另外配置的，则用户必须设置应答头。在这种情况下，在初始化过程中会写入 INFO 日志消息。下面的示例使用 KafkaHeaders.REPLY_TOPIC :

record.headers().add(new RecordHeader(KafkaHeaders.REPLY_TOPIC, "kReplies".getBytes()));

在配置单个回复 TopicPartitionOffset 时，只要每个实例侦听不同的分区，就可以为多个模板使用相同的回复主题。在配置单个回复主题时，每个实例必须使用不同的 group.id 。在这种情况下，所有实例都会接收每个答复，但只有发送请求的实例才会找到相关 ID。这对于自动缩放可能是有用的，但需要额外的网络流量开销，并且丢弃每个不需要的回复的成本很小。使用此设置时，我们建议你将模板的 sharedReplyTopic 设置为 true ，这将减少对调试的意外回复的日志级别，而不是默认错误。

下面是一个配置应答容器以使用相同的共享应答主题的示例：

@Bean
public ConcurrentMessageListenerContainer<String, String> replyContainer(
        ConcurrentKafkaListenerContainerFactory<String, String> containerFactory) {
    ConcurrentMessageListenerContainer<String, String> container = containerFactory.createContainer("topic2");
    container.getContainerProperties().setGroupId(UUID.randomUUID().toString()); // unique
    Properties props = new Properties();
    props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest"); // so the new group doesn't get old replies
    container.getContainerProperties().setKafkaConsumerProperties(props);
    return container;

默认情况下，使用 3 个标题：

• KafkaHeaders.CORRELATION_ID -用于将回复与请求关联起来

• KafkaHeaders.REPLY_TOPIC -用于告诉服务器在哪里回复

• KafkaHeaders.REPLY_PARTITION -（可选）用于告诉服务器要回复哪个分区

@KafkaListener 基础架构使用这些头名称来路由答复。

从版本 2.3 开始，你可以自定义标题名称-模板有 3 个属性 correlationHeaderName 、 replyTopicHeaderName 和 replyPartitionHeaderName 。如果你的服务器不是 Spring 应用程序（或者不使用 @KafkaListener ），这是有用的。

# 请求/回复 Message<?> s

版本 2.7 在 ReplyingKafkaTemplate 中添加了发送和接收 spring-messaging 的 Message<?> 抽象的方法：

RequestReplyMessageFuture<K, V> sendAndReceive(Message<?> message);
<P> RequestReplyTypedMessageFuture<K, V, P> sendAndReceive(Message<?> message,
        ParameterizedTypeReference<P> returnType);

这些将使用模板的默认 replyTimeout ，也有重载版本可以在方法调用中占用超时时间。

如果使用者的 Deserializer 或模板的 MessageConverter 可以通过配置或在回复消息中键入元数据来转换有效负载，而不需要任何其他信息，请使用第一种方法。

如果需要为返回类型提供类型信息，请使用第二种方法来帮助消息转换器。这还允许相同的模板接收不同的类型，即使在答复中没有类型元数据，例如当服务器端不是 Spring 应用程序时也是如此。以下是后者的一个例子：

例 6.模板 Bean

Java

@Bean
ReplyingKafkaTemplate<String, String, String> template(
        ProducerFactory<String, String> pf,
        ConcurrentKafkaListenerContainerFactory<String, String> factory) {
    ConcurrentMessageListenerContainer<String, String> replyContainer =
            factory.createContainer("replies");
    replyContainer.getContainerProperties().setGroupId("request.replies");
    ReplyingKafkaTemplate<String, String, String> template =
            new ReplyingKafkaTemplate<>(pf, replyContainer);
    template.setMessageConverter(new ByteArrayJsonMessageConverter());
    template.setDefaultTopic("requests");
    return template;

Kotlin

@Bean
fun template(
    pf: ProducerFactory<String?, String>?,
    factory: ConcurrentKafkaListenerContainerFactory<String?, String?>
): ReplyingKafkaTemplate<String?, String, String?> {
    val replyContainer = factory.createContainer("replies")
    replyContainer.containerProperties.groupId = "request.replies"
    val template = ReplyingKafkaTemplate(pf, replyContainer)
    template.messageConverter = ByteArrayJsonMessageConverter()
    template.defaultTopic = "requests"
    return template

例 7.使用模板

Java

RequestReplyTypedMessageFuture<String, String, Thing> future1 =
        template.sendAndReceive(MessageBuilder.withPayload("getAThing").build(),
                new ParameterizedTypeReference<Thing>() { });
log.info(future1.getSendFuture().get(10, TimeUnit.SECONDS).getRecordMetadata().toString());
Thing thing = future1.get(10, TimeUnit.SECONDS).getPayload();
log.info(thing.toString());
RequestReplyTypedMessageFuture<String, String, List<Thing>> future2 =
        template.sendAndReceive(MessageBuilder.withPayload("getThings").build(),
                new ParameterizedTypeReference<List<Thing>>() { });
log.info(future2.getSendFuture().get(10, TimeUnit.SECONDS).getRecordMetadata().toString());
List<Thing> things = future2.get(10, TimeUnit.SECONDS).getPayload();
things.forEach(thing1 -> log.info(thing1.toString()));

Kotlin

val future1: RequestReplyTypedMessageFuture<String?, String?, Thing?>? =
    template.sendAndReceive(MessageBuilder.withPayload("getAThing").build(),
        object : ParameterizedTypeReference<Thing?>() {})
log.info(future1?.sendFuture?.get(10, TimeUnit.SECONDS)?.recordMetadata?.toString())
val thing = future1?.get(10, TimeUnit.SECONDS)?.payload
log.info(thing.toString())
val future2: RequestReplyTypedMessageFuture<String?, String?, List<Thing?>?>? =
    template.sendAndReceive(MessageBuilder.withPayload("getThings").build(),
        object : ParameterizedTypeReference<List<Thing?>?>() {})
log.info(future2?.sendFuture?.get(10, TimeUnit.SECONDS)?.recordMetadata.toString())
val things = future2?.get(10, TimeUnit.SECONDS)?.payload
things?.forEach(Consumer { thing1: Thing? -> log.info(thing1.toString()) })

# 回复类型消息 <?>

当 @KafkaListener 返回 Message<?> 时，在版本为 2.5 之前的情况下，需要填充回复主题和相关 ID 头。在本例中，我们使用请求中的回复主题标头：

@KafkaListener(id = "requestor", topics = "request")
@SendTo
public Message<?> messageReturn(String in) {
    return MessageBuilder.withPayload(in.toUpperCase())
            .setHeader(KafkaHeaders.TOPIC, replyTo)
            .setHeader(KafkaHeaders.MESSAGE_KEY, 42)
            .setHeader(KafkaHeaders.CORRELATION_ID, correlation)
            .build();

这也显示了如何在回复记录上设置一个键。

从版本 2.5 开始，该框架将检测这些标题是否丢失，并用主题填充它们-从 @SendTo 值确定的主题或传入的 KafkaHeaders.REPLY_TOPIC 标题（如果存在）。如果存在，它还将响应传入的 KafkaHeaders.CORRELATION_ID 和 KafkaHeaders.REPLY_PARTITION 。

@KafkaListener(id = "requestor", topics = "request")
@SendTo  // default REPLY_TOPIC header
public Message<?> messageReturn(String in) {
    return MessageBuilder.withPayload(in.toUpperCase())
            .setHeader(KafkaHeaders.MESSAGE_KEY, 42)
            .build();

# 聚合多个回复

[使用 ReplyingKafkaTemplate ]（#replying-template）中的模板严格用于单个请求/回复场景。对于单个消息的多个接收者返回答复的情况，可以使用 AggregatingReplyingKafkaTemplate 。这是 散-集 Enterprise 集成模式 (opens new window) 客户端的一个实现。

与 ReplyingKafkaTemplate 类似， AggregatingReplyingKafkaTemplate 构造函数需要一个生产者工厂和一个侦听器容器来接收回复；它有第三个参数 BiPredicate<List<ConsumerRecord<K, R>>, Boolean> releaseStrategy ，在每次接收到回复时都会查询这个参数；当谓词返回 true 时， ConsumerRecord s 的集合用于完成由 sendAndReceive 方法返回的 Future 。

还有一个额外的属性 returnPartialOnTimeout （默认为 false）。当这被设置为 true 时，而不是用 KafkaReplyTimeoutException 来完成 future，部分结果通常会完成 future（只要至少收到了一条回复记录）。

从版本 2.3.5 开始，在超时之后也调用谓词（如果 returnPartialOnTimeout 是 true ）。第一个参数是当前的记录列表；第二个参数是 true ，如果这个调用是由于超时引起的。谓词可以修改记录列表。

AggregatingReplyingKafkaTemplate<Integer, String, String> template =
        new AggregatingReplyingKafkaTemplate<>(producerFactory, container,
                        coll -> coll.size() == releaseSize);
RequestReplyFuture<Integer, String, Collection<ConsumerRecord<Integer, String>>> future =
        template.sendAndReceive(record);
future.getSendFuture().get(10, TimeUnit.SECONDS); // send ok
ConsumerRecord<Integer, Collection<ConsumerRecord<Integer, String>>> consumerRecord =
        future.get(30, TimeUnit.SECONDS);

请注意，返回类型是 ConsumerRecord ，其值是 ConsumerRecord s 的集合。该“外” ConsumerRecord 不是一个“真实”的记录，它是由模板合成的，作为实际接收到的回复记录的持有者用于请求。当正常的发布发生时（Release Strategy 返回 true），主题设置为 aggregatedResults ；如果 returnPartialOnTimeout 为真，并且发生超时（并且至少收到了一条回复记录），主题设置为 partialResultsAfterTimeout 。模板为这些“主题”名称提供了常量静态变量：

/**
 * Pseudo topic name for the "outer" {@link ConsumerRecords} that has the aggregated
 * results in its value after a normal release by the release strategy.
public static final String AGGREGATED_RESULTS_TOPIC = "aggregatedResults";
 * Pseudo topic name for the "outer" {@link ConsumerRecords} that has the aggregated
 * results in its value after a timeout.
public static final String PARTIAL_RESULTS_AFTER_TIMEOUT_TOPIC = "partialResultsAfterTimeout";

在 Collection 中，真正的 ConsumerRecord 包含接收答复的实际主题。

# 4.1.4.接收消息

可以通过配置 MessageListenerContainer 并提供消息侦听器或使用 @KafkaListener 注释来接收消息。

# 消息侦听器

当使用 消息侦听器容器 时，必须提供一个侦听器来接收数据。目前，消息侦听器有八个受支持的接口。下面的清单展示了这些接口：

public interface MessageListener<K, V> { (1)
    void onMessage(ConsumerRecord<K, V> data);
public interface AcknowledgingMessageListener<K, V> { (2)
    void onMessage(ConsumerRecord<K, V> data, Acknowledgment acknowledgment);
public interface ConsumerAwareMessageListener<K, V> extends MessageListener<K, V> { (3)
    void onMessage(ConsumerRecord<K, V> data, Consumer<?, ?> consumer);
public interface AcknowledgingConsumerAwareMessageListener<K, V> extends MessageListener<K, V> { (4)
    void onMessage(ConsumerRecord<K, V> data, Acknowledgment acknowledgment, Consumer<?, ?> consumer);
public interface BatchMessageListener<K, V> { (5)
    void onMessage(List<ConsumerRecord<K, V>> data);
public interface BatchAcknowledgingMessageListener<K, V> { (6)
    void onMessage(List<ConsumerRecord<K, V>> data, Acknowledgment acknowledgment);
public interface BatchConsumerAwareMessageListener<K, V> extends BatchMessageListener<K, V> { (7)
    void onMessage(List<ConsumerRecord<K, V>> data, Consumer<?, ?> consumer);
public interface BatchAcknowledgingConsumerAwareMessageListener<K, V> extends BatchMessageListener<K, V> { (8)
    void onMessage(List<ConsumerRecord<K, V>> data, Acknowledgment acknowledgment, Consumer<?, ?> consumer);

# 消息侦听器容器

提供了两个 MessageListenerContainer 实现：

• KafkaMessageListenerContainer

• ConcurrentMessageListenerContainer

KafkaMessageListenerContainer 接收来自单个线程上所有主题或分区的所有消息。 ConcurrentMessageListenerContainer 将委托给一个或多个 KafkaMessageListenerContainer 实例，以提供多线程消耗。

从版本 2.2.7 开始，你可以将 RecordInterceptor 添加到侦听器容器；在调用侦听器允许检查或修改记录之前，将调用它。如果拦截器返回 null，则不调用侦听器。从版本 2.7 开始，它有额外的方法，在侦听器退出后调用这些方法（通常是通过抛出异常）。此外，从版本 2.7 开始，现在有一个 BatchInterceptor ，为 批处理侦听器 提供类似的功能。此外， ConsumerAwareRecordInterceptor （和 BatchInterceptor ）提供对 Consumer<?, ?> 的访问。例如，这可以用来访问拦截器中的消费者指标。

CompositeRecordInterceptor 和 CompositeBatchInterceptor 可用于调用多个拦截器。

默认情况下，从版本 2.8 开始，当使用事务时，拦截器在事务启动之前被调用。你可以将侦听器容器的 interceptBeforeTx 属性设置为 false ，以便在事务启动后调用拦截器。

从版本 2.3.8、2.4.6 开始，当并发性大于 1 时， ConcurrentMessageListenerContainer 现在支持 静态成员 (opens new window) 。 group.instance.id 后缀为 -n ，后缀为 n ，起始于 1 。这与增加的 session.timeout.ms 一起，可以用来减少重新平衡事件，例如，当应用程序实例重新启动时。

# 使用 KafkaMessageListenerContainer

以下构造函数可用：

public KafkaMessageListenerContainer(ConsumerFactory<K, V> consumerFactory,
                    ContainerProperties containerProperties)

它在 ContainerProperties 对象中接收 ConsumerFactory 和有关主题和分区以及其他配置的信息。 ContainerProperties 具有以下构造函数：

public ContainerProperties(TopicPartitionOffset... topicPartitions)
public ContainerProperties(String... topics)
public ContainerProperties(Pattern topicPattern)

第一个构造函数接受一个由 TopicPartitionOffset 参数组成的数组，以显式地指示容器使用哪些分区（使用 Consumer assign() 方法），并使用一个可选的初始偏移量。在默认情况下，正值是绝对的偏移量。在默认情况下，负值是相对于分区中当前的最后一个偏移量的。为 TopicPartitionOffset 提供了一个构造函数，它接受一个额外的 boolean 参数。如果这是 true ，则初始偏移（正或负）相对于此消费者的当前位置。当容器启动时，将应用这些偏移量。第二个是一个主题数组，Kafka 基于 group.id 属性（在整个组中分发分区）分配分区。第三种使用 regex Pattern 来选择主题。

要将 MessageListener 分配给容器，可以在创建容器时使用 ContainerProps.setMessageListener 方法。下面的示例展示了如何做到这一点：

ContainerProperties containerProps = new ContainerProperties("topic1", "topic2");
containerProps.setMessageListener(new MessageListener<Integer, String>() {
DefaultKafkaConsumerFactory<Integer, String> cf =
                        new DefaultKafkaConsumerFactory<>(consumerProps());
KafkaMessageListenerContainer<Integer, String> container =
                        new KafkaMessageListenerContainer<>(cf, containerProps);
return container;

请注意，当创建 DefaultKafkaConsumerFactory 时，使用只接收上述属性的构造函数意味着从配置中提取键和值 Deserializer 类。或者， Deserializer 实例可以传递给 DefaultKafkaConsumerFactory 构造函数，用于键和/或值，在这种情况下，所有消费者共享相同的实例。另一种选择是提供 Supplier<Deserializer> s（从版本 2.3 开始），用于为每个 Consumer 获取单独的 Deserializer 实例：

DefaultKafkaConsumerFactory<Integer, CustomValue> cf =
                        new DefaultKafkaConsumerFactory<>(consumerProps(), null, () -> new CustomValueDeserializer());
KafkaMessageListenerContainer<Integer, String> container =
                        new KafkaMessageListenerContainer<>(cf, containerProps);
return container;

有关可以设置的各种属性的更多信息，请参见 Javadoc (opens new window) for ContainerProperties 。

自版本 2.1.1 以来，一个名为 logContainerConfig 的新属性可用。当启用 true 和 INFO 日志记录时，每个侦听器容器写一个日志消息，总结其配置属性。

默认情况下，主题偏移提交的日志记录是在 DEBUG 日志级别执行的。从版本 2.1.2 开始， ContainerProperties 中的一个名为 commitLogLevel 的属性允许你为这些消息指定日志级别。例如，要将日志级别更改为 INFO ，可以使用 containerProperties.setCommitLogLevel(LogIfLevelEnabled.Level.INFO); 。

从版本 2.2 开始，添加了一个名为 missingTopicsFatal 的新容器属性（默认值： false 自 2.3.4 起）。如果代理上不存在任何已配置的主题，这将阻止容器启动。如果容器被配置为侦听主题模式（regex），则不会应用该选项。以前，容器线程在 consumer.poll() 方法中循环运行，等待在记录许多消息时出现主题。除了日志之外，没有迹象表明存在问题。

从版本 2.8 开始，引入了一个新的容器属性 authExceptionRetryInterval 。这将导致容器在从 KafkaConsumer 获取任何 AuthenticationException 或 AuthorizationException 后重试获取消息。例如，当被配置的用户被拒绝读取某个主题或凭据不正确时，就会发生这种情况。定义 authExceptionRetryInterval 允许容器在授予适当权限时恢复。

从版本 2.8 开始，在创建消费者工厂时，如果你将反序列化器作为对象（在构造函数中或通过 setter）提供，工厂将调用 configure() 方法来使用配置属性对它们进行配置。

# 使用 ConcurrentMessageListenerContainer

单个构造函数类似于 KafkaListenerContainer 构造函数。下面的清单显示了构造函数的签名：

public ConcurrentMessageListenerContainer(ConsumerFactory<K, V> consumerFactory,
                            ContainerProperties containerProperties)

它还具有 concurrency 属性。例如， container.setConcurrency(3) 创建了三个 KafkaMessageListenerContainer 实例。

对于第一个构造函数，Kafka 使用其组管理功能在消费者之间分配分区。

当容器属性配置为 TopicPartitionOffset s 时， ConcurrentMessageListenerContainer 将 TopicPartitionOffset 实例分布在委托 KafkaMessageListenerContainer 实例中。

假设提供了六个 TopicPartitionOffset 实例，并且 concurrency 是 3 ；每个容器都有两个分区。对于五个 TopicPartitionOffset 实例，两个容器获得两个分区，第三个容器获得一个分区。如果 concurrency 大于 TopicPartitions 的个数，则对 concurrency 进行向下调整，以便每个容器获得一个分区。

从版本 1.3 开始， MessageListenerContainer 提供对底层 KafkaConsumer 的度量的访问。在 ConcurrentMessageListenerContainer 的情况下， metrics() 方法返回所有目标 KafkaMessageListenerContainer 实例的度量。度量值由为底层 KafkaConsumer 提供的 client-id 分组为 Map<MetricName, ? extends Metric> 。

从版本 2.3 开始， ContainerProperties 提供了一个 idleBetweenPolls 选项，让侦听器容器中的主循环在 KafkaConsumer.poll() 调用之间休眠。从所提供的选项和 max.poll.interval.ms 消费者配置和当前记录批处理时间之间的差值中选择一个实际的睡眠间隔作为最小值。

# 提交偏移

为提交偏移提供了几个选项。如果 enable.auto.commit 消费者属性是 true ，Kafka 将根据其配置自动提交偏移。如果是 false ，则容器支持几个 AckMode 设置（在下一个列表中进行了描述）。默认的 AckMode 是 BATCH 。从版本 2.3 开始，该框架将 enable.auto.commit 设置为 false ，除非在配置中明确设置。以前，如果未设置属性，则使用 Kafka 默认值（ true ）。

消费者 poll() 方法返回一个或多个 ConsumerRecords 。为每个记录调用 MessageListener 。下面的列表描述了容器为每个 AckMode （不使用事务时）所采取的操作：

• RECORD :在侦听器在处理完记录后返回时提交偏移量。

• BATCH :在处理完 poll() 返回的所有记录后提交偏移量。

• TIME :在 poll() 返回的所有记录都已被处理的情况下提交偏移量，只要 ackTime 自上次提交以来的偏移量已被超过。

• COUNT :提交当 poll() 返回的所有记录都已被处理时的偏移量，只要 ackCount 记录自上次提交以来一直被接收。

• COUNT_TIME :类似于 TIME 和 COUNT ，但如果任一条件是 true ，则执行提交。

• MANUAL :消息侦听器负责 acknowledge() 的 Acknowledgment 。在此之后，将应用与 BATCH 相同的语义。

• MANUAL_IMMEDIATE :当侦听器调用 Acknowledgment.acknowledge() 方法时，立即提交偏移量。

当使用 交易 时，偏移量被发送到事务，语义等价于 RECORD 或 BATCH ，这取决于侦听器类型（记录或批处理）。

根据 syncCommits 容器属性，将使用消费者上的 commitSync() 或 commitAsync() 方法。 syncCommits 默认情况下是 true ；还请参见 setSyncCommitTimeout 。参见 setCommitCallback 以获取异步提交的结果；默认的回调是 LoggingCommitCallback ，它记录错误（和调试级别的成功）。

因为侦听器容器有自己的提交偏移的机制，所以它更喜欢 kafka ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG 为 false 。从版本 2.3 开始，它无条件地将其设置为 false，除非在消费者工厂或容器的消费者属性重写中专门设置了 false。

Acknowledgment 具有以下方法：

public interface Acknowledgment {
    void acknowledge();

此方法使侦听器能够控制何时提交偏移。

从版本 2.3 开始， Acknowledgment 接口有两个额外的方法 nack(long sleep) 和 nack(int index, long sleep) 。第一个用于记录侦听器，第二个用于批处理侦听器。为侦听器类型调用错误的方法将抛出 IllegalStateException 。

对于记录侦听器，当调用 nack() 时，将提交任何挂起的偏移量，丢弃上一次轮询的重置记录，并在它们的分区上执行查找，以便在下一次 poll() 上重新交付失败的记录和未处理的记录。通过设置 sleep 参数，可以在重新交付之前暂停使用者线程。这类似于当容器配置为 DefaultErrorHandler 时抛出异常的功能。

使用批处理侦听器时，可以在发生故障的批处理中指定索引。当调用 nack() 时，将对记录提交偏移，然后在分区上对失败和丢弃的记录执行索引和查找，以便在下一个 poll() 上重新交付它们。

有关更多信息，请参见 容器错误处理程序 。

# 侦听器容器自动启动

侦听器容器实现 SmartLifecycle ，而 autoStartup 默认情况下是 true 。容器在后期启动（ Integer.MAX-VALUE - 100 ）。实现 SmartLifecycle 以处理来自侦听器的数据的其他组件应该在较早的阶段启动。 - 100 为后面的阶段留出了空间，以使组件能够在容器之后自动启动。

# 手动提交偏移

通常，当使用 AckMode.MANUAL 或 AckMode.MANUAL_IMMEDIATE 时，必须按顺序确认确认，因为 Kafka 不为每个记录维护状态，只为每个组/分区维护一个提交的偏移量。从版本 2.8 开始，你现在可以设置容器属性 asyncAcks ，它允许以任何顺序确认投票返回的记录的确认。侦听器容器将推迟顺序外的提交，直到收到缺少的确认。消费者将被暂停（没有新的记录交付），直到前一次投票的所有补偿都已提交。

# @KafkaListener 注释

@KafkaListener 注释用于指定 Bean 方法作为侦听器容器的侦听器。 Bean 包装在 MessagingMessageListenerAdapter 中配置有各种特征，例如转换器来转换数据，如果需要，以匹配该方法的参数。

可以使用 #{…​} 或属性占位符（ ${…​} ）使用 SPEL 配置注释上的大多数属性。有关更多信息，请参见 Javadoc (opens new window) 。

# 记录收听者

@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(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokersAsString());
        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 表达式；有关如何生成一个大的分区列表的示例，请参见[[tip-assign-all-parts]。

从版本 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 个分区。

# 手动确认

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

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

# 消费者记录元数据

最后，关于记录的元数据可以从消息头获得。你可以使用以下头名称来检索消息的头：

• KafkaHeaders.OFFSET

• KafkaHeaders.RECEIVED_MESSAGE_KEY

• KafkaHeaders.RECEIVED_TOPIC

• KafkaHeaders.RECEIVED_PARTITION_ID

• KafkaHeaders.RECEIVED_TIMESTAMP

• KafkaHeaders.TIMESTAMP_TYPE

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

下面的示例展示了如何使用标题：

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

从版本 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;

下面的示例展示了如何接收有效载荷列表：

@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_MESSAGE_KEY) List<Integer> keys,
        @Header(KafkaHeaders.RECEIVED_PARTITION_ID) 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 listen14(List<Message<?>> list) {
@KafkaListener(id = "listMsgAck", topics = "myTopic", containerFactory = "batchFactory")
public void listen15(List<Message<?>> list, Acknowledgment ack) {
@KafkaListener(id = "listMsgAckConsumer", topics = "myTopic", containerFactory = "batchFactory")
public void listen16(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) {

# 注释属性

从版本 2.0 开始， id 属性（如果存在）被用作 Kafka Consumer group.id 属性，如果存在，则覆盖 Consumer 工厂中的配置属性。还可以显式地将 groupId 设置为 idIsGroup ，也可以将 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;

如果在不太可能的情况下，你有一个实际的 Bean 名为 __listener ，那么你可以使用 beanRef 属性来更改表达式标记。下面的示例展示了如何做到这一点：

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

从版本 2.2.4 开始，你可以直接在注释中指定 Kafka 消费者属性，这些属性将覆盖在消费者工厂中配置的具有相同名称的任何属性。以这种方式指定 不能 和 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 ]（#routing-template）示例中的相应侦听器的示例。

@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));

# 获取消费者 group.id

当在多个容器中运行相同的侦听器代码时，能够确定记录来自哪个容器（由其 group.id 消费者属性标识）可能是有用的。

你可以在侦听器线程上调用 KafkaUtils.getConsumerGroupId() 来执行此操作。或者，你可以访问方法参数中的组 ID。

@KafkaListener(id = "bar", topicPattern = "${topicTwo:annotated2}", exposeGroupId = "${always:true}")
public void listener(@Payload String foo,
        @Header(KafkaHeaders.GROUP_ID) String groupId) {

# 容器线程命名

侦听器容器当前使用两个任务执行器，一个用于调用使用者，另一个用于在 Kafka 消费者属性 enable.auto.commit 为 false 时调用侦听器。你可以通过设置容器的 consumerExecutor 和 listenerExecutor 属性来提供自定义执行器。当使用池执行程序时，确保有足够多的线程可用来处理使用它们的所有容器之间的并发性。当使用 ConcurrentMessageListenerContainer 时，来自每个使用者的线程都用于每个使用者（ concurrency ）。

如果不提供消费者执行器，则使用 SimpleAsyncTaskExecutor 。此执行器创建名称与 <beanName>-C-1 （使用者线程）类似的线程。对于 ConcurrentMessageListenerContainer ，线程名称的 <beanName> 部分变成 <beanName>-m ，其中 m 表示消费者实例。 n 每次启动容器时都会增加。所以，具有 Bean 名称的 container ，此容器中的线程将被命名为 container-0-C-1 、 container-1-C-1 等，在容器被第一次启动之后； container-0-C-2 、 container-1-C-2 等，在停止之后又被随后的启动。

# @KafkaListener 作为元注释

从版本 2.2 开始，你现在可以使用 @KafkaListener 作为元注释。下面的示例展示了如何做到这一点：

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@KafkaListener
public @interface MyThreeConsumersListener {
    @AliasFor(annotation = KafkaListener.class, attribute = "id")
    String id();
    @AliasFor(annotation = KafkaListener.class, attribute = "topics")
    String[] topics();
    @AliasFor(annotation = KafkaListener.class, attribute = "concurrency")
    String concurrency() default "3";

你必须至少别名 topics 、 topicPattern 或 topicPartitions 中的一个（并且，通常是 id 或 groupId ，除非你在消费者工厂配置中指定了 group.id ）。下面的示例展示了如何做到这一点：

@MyThreeConsumersListener(id = "my.group", topics = "my.topic")
public void listen1(String in) {

# 在类上 @KafkaListener

在类级别上使用 @KafkaListener 时，必须在方法级别上指定 @KafkaHandler 。在发送消息时，将使用转换后的消息有效负载类型来确定调用哪个方法。下面的示例展示了如何做到这一点：

@KafkaListener(id = "multi", topics = "myTopic")
static class MultiListenerBean {
    @KafkaHandler
    public void listen(String foo) {
    @KafkaHandler
    public void listen(Integer bar) {
    @KafkaHandler(isDefault = true)
    public void listenDefault(Object object) {

从版本 2.1.3 开始，你可以将 @KafkaHandler 方法指定为默认方法，如果其他方法不匹配，则调用该方法。最多只能指定一种方法。当使用 @KafkaHandler 方法时，有效负载必须已经转换为域对象（因此可以执行匹配）。使用自定义的反序列化器， JsonDeserializer ，或 JsonMessageConverter ，其 TypePrecedence 设置为 TYPE_ID 。有关更多信息，请参见 序列化、反序列化和消息转换 。

例如：

@KafkaHandler(isDefault = true)
public void listenDefault(Object object, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {

如果对象是 String ，这将不起作用； topic 参数还将获得对 object 的引用。

如果在默认方法中需要有关记录的元数据，请使用以下方法：

@KafkaHandler(isDefault = true)
void listen(Object in, @Header(KafkaHeaders.RECORD_METADATA) ConsumerRecordMetadata meta) {
    String topic = meta.topic();

# topic 属性修改

从版本 2.7.2 开始，你现在可以在创建容器之前以编程方式修改注释属性。为此，将一个或多个 KafkaListenerAnnotationBeanPostProcessor.AnnotationEnhancer 添加到应用程序上下文。 AnnotationEnhancer 是一个 BiFunction<Map<String, Object>, AnnotatedElement, Map<String, Object> ，并且必须返回属性映射。属性值可以包含 SPEL 和/或属性占位符；在执行任何解析之前都会调用增强器。如果存在多个增强器，并且它们实现 Ordered ，则将按顺序调用它们。

以下是一个例子：

@Bean
public static AnnotationEnhancer groupIdEnhancer() {
    return (attrs, element) -> {
        attrs.put("groupId", attrs.get("id") + "." + (element instanceof Class
                ? ((Class<?>) element).getSimpleName()
                : ((Method) element).getDeclaringClass().getSimpleName()
                        +  "." + ((Method) element).getName()));
        return attrs;

# @KafkaListener 生命周期管理

为 @KafkaListener 注释创建的侦听器容器不是应用程序上下文中的 bean。相反，它们被注册在类型 KafkaListenerEndpointRegistry 的基础结构 Bean 中。 Bean 由框架自动声明并管理容器的生命周期；它将自动启动将 autoStartup 设置为 true 的任何容器。由所有容器工厂创建的所有容器必须在相同的 phase 中。有关更多信息，请参见 监听器容器自动启动 。你可以通过使用注册表以编程方式管理生命周期。启动或停止注册表将启动或停止所有已注册的容器。或者，你可以通过使用其 id 属性获得对单个容器的引用。你可以在注释上设置 autoStartup ，这会覆盖配置到容器工厂中的默认设置。你可以从应用程序上下文中获得对 Bean 的引用，例如自动布线，以管理其注册的容器。下面的例子说明了如何做到这一点：

@KafkaListener(id = "myContainer", topics = "myTopic", autoStartup = "false")
public void listen(...) { ... }

@Autowired
private KafkaListenerEndpointRegistry registry;
    this.registry.getListenerContainer("myContainer").start();

注册中心仅维护其管理的容器的生命周期；声明为 bean 的容器不受注册中心的管理，可以从应用程序上下文中获得。可以通过调用注册表的 getListenerContainers() 方法获得托管容器的集合。版本 2.2.5 添加了一个方便的方法 getAllListenerContainers() ，该方法返回所有容器的集合，包括由注册中心管理的容器和声明为 bean 的容器。返回的集合将包括任何已初始化的原型 bean，但它不会初始化任何懒惰的 Bean 声明。

# @KafkaListener``@Payload 验证

从版本 2.2 开始，现在更容易添加 Validator 来验证 @KafkaListener``@Payload 参数。以前，你必须配置一个自定义 DefaultMessageHandlerMethodFactory 并将其添加到注册商。现在，你可以将验证器添加到注册器本身。下面的代码展示了如何做到这一点：

@Configuration
@EnableKafka
public class Config implements KafkaListenerConfigurer {
    @Override
    public void configureKafkaListeners(KafkaListenerEndpointRegistrar registrar) {
      registrar.setValidator(new MyValidator());

@Configuration
@EnableKafka
public class Config implements KafkaListenerConfigurer {
    @Autowired
    private LocalValidatorFactoryBean validator;
    @Override
    public void configureKafkaListeners(KafkaListenerEndpointRegistrar registrar) {
      registrar.setValidator(this.validator);

以下示例展示了如何验证：

public static class ValidatedClass {
  @Max(10)
  private int bar;
  public int getBar() {
    return this.bar;
  public void setBar(int bar) {
    this.bar = bar;

@KafkaListener(id="validated", topics = "annotated35", errorHandler = "validationErrorHandler",
      containerFactory = "kafkaJsonListenerContainerFactory")
public void validatedListener(@Payload @Valid ValidatedClass val) {
@Bean
public KafkaListenerErrorHandler validationErrorHandler() {
    return (m, e) -> {

从版本 2.5.11 开始，验证现在可以在类级侦听器中的 KafkaMessageListenerContainer 方法的有效负载上进行。参见[ @KafkaListener on a class]（#class-level-kafkalistener）。

# 重新平衡听众

ContainerProperties 具有一个名为 consumerRebalanceListener 的属性，它接受了 Kafka 客户机的 ConsumerRebalanceListener 接口的一个实现。如果不提供此属性，则容器将配置一个日志侦听器，该侦听器将在 INFO 级别记录重新平衡事件。该框架还添加了一个子接口 @KafkaListener 。下面的清单显示了 ConsumerAwareRebalanceListener 接口定义：

public interface ConsumerAwareRebalanceListener extends ConsumerRebalanceListener {
    void onPartitionsRevokedBeforeCommit(Consumer<?, ?> consumer, Collection<TopicPartition> partitions);
    void onPartitionsRevokedAfterCommit(Consumer<?, ?> consumer, Collection<TopicPartition> partitions);
    void onPartitionsAssigned(Consumer<?, ?> consumer, Collection<TopicPartition> partitions);
    void onPartitionsLost(Consumer<?, ?> consumer, Collection<TopicPartition> partitions);

注意，当分区被撤销时有两个回调。第一个是立即调用的。第二种方法是在任何未完成的补偿被提交后调用。如果你希望在某些外部存储库中维护偏移，这是非常有用的，如下例所示：

containerProperties.setConsumerRebalanceListener(new ConsumerAwareRebalanceListener() {
    @Override
    public void onPartitionsRevokedBeforeCommit(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
        // acknowledge any pending Acknowledgments (if using manual acks)
    @Override
    public void onPartitionsRevokedAfterCommit(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
        // ...
            store(consumer.position(partition));
        // ...
    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // ...
            consumer.seek(partition, offsetTracker.getOffset() + 1);
        // ...

# 使用 @SendTo 转发监听器结果

从版本 2.0 开始，如果你还使用 @KafkaListener 注释 @KafkaListener ，并且方法调用返回一个结果，则结果将被转发到 一次语义学 指定的主题。

@SendTo 值可以有几种形式：

• @SendTo("someTopic") 路由到字面主题

• KafkaTemplate 路由到主题，该主题是在应用程序上下文初始化期间通过计算表达式一次来确定的。

• @SendTo("!{someExpression}") 路由到通过在运行时计算表达式来确定的主题。求值的 #root 对象具有三个属性：

  • request :入站 ConsumerRecord （或用于批处理侦听器的 ConsumerRecords 对象）

  • source :从 request 转换而来的 org.springframework.messaging.Message<?> 。

  • result :方法返回结果。

• @SendTo （没有属性）：这被视为 !{source.headers['kafka_replyTopic']} （自版本 2.1.3）。

从版本 2.1.11 和 2.2.1 开始，属性占位符在 @SendTo 值内解析。

表达式求值的结果必须是表示主题名称的 String 。以下示例展示了使用 @SendTo 的各种方法：

@KafkaListener(topics = "annotated21")
@SendTo("!{request.value()}") // runtime SpEL
public String replyingListener(String in) {
@KafkaListener(topics = "${some.property:annotated22}")
@SendTo("#{myBean.replyTopic}") // config time SpEL
public Collection<String> replyingBatchListener(List<String> in) {
@KafkaListener(topics = "annotated23", errorHandler = "replyErrorHandler")
@SendTo("annotated23reply") // static reply topic definition
public String replyingListenerWithErrorHandler(String in) {
@KafkaListener(topics = "annotated25")
@SendTo("annotated25reply1")
public class MultiListenerSendTo {
    @KafkaHandler
    public String foo(String in) {
    @KafkaHandler
    @SendTo("!{'annotated25reply2'}")
    public String bar(@Payload(required = false) KafkaNull nul,
            @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) int key) {

从版本 2.2 开始，你可以向监听器容器工厂添加 ReplyHeadersConfigurer 。查询此项以确定你想要在回复消息中设置哪些头。下面的示例展示了如何添加 ReplyHeadersConfigurer :

@Bean
public ConcurrentKafkaListenerContainerFactory<Integer, String> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(cf());
    factory.setReplyTemplate(template());
    factory.setReplyHeadersConfigurer((k, v) -> k.equals("cat"));
    return factory;

如果你愿意，还可以添加更多的标题。下面的示例展示了如何做到这一点：

@Bean
public ConcurrentKafkaListenerContainerFactory<Integer, String> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(cf());
    factory.setReplyTemplate(template());
    factory.setReplyHeadersConfigurer(new ReplyHeadersConfigurer() {
      @Override
      public boolean shouldCopy(String headerName, Object headerValue) {
        return false;
      @Override
      public Map<String, Object> additionalHeaders() {
        return Collections.singletonMap("qux", "fiz");
    return factory;

当使用 @SendTo 时，必须在其 replyTemplate 属性中配置 ReplyHeadersConfigurer ，以执行发送。

@Bean
public KafkaTemplate<String, String> myReplyingTemplate() {
    return new KafkaTemplate<Integer, String>(producerFactory()) {
        @Override
        public ListenableFuture<SendResult<String, String>> send(String topic, String data) {
            return super.send(topic, partitionForData(data), keyForData(data), data);

当使用请求/回复语义时，目标分区可以由发送方请求。

# 过滤消息

在某些情况下，例如重新平衡，已经处理过的消息可能会被重新传递。框架不能知道这样的消息是否已被处理。这是一个应用程序级函数。这被称为 幂等接收机 (opens new window) 模式，并且 Spring 集成提供了 幂等接收机 (opens new window) 。

Spring for Apache Kafka 项目还通过 FilteringMessageListenerAdapter 类提供了一些帮助，它可以包装你的 MessageListener 。该类接受 RecordFilterStrategy 的实现，在该实现中，你实现 filter 方法，以表示消息是重复的，应该丢弃。这有一个名为 ackDiscarded 的附加属性，它指示适配器是否应该确认丢弃的记录。默认情况下是 false 。

当使用 @KafkaListener 时，在容器工厂上设置 RecordFilterStrategy （以及可选的 ackDiscarded ），以便侦听器被包装在适当的过滤适配器中。

此外，当你使用批处理 消息监听器 时，还提供了一个 FilteringBatchMessageListenerAdapter 。

# 重试送货

参见 处理异常 中的 DefaultErrorHandler 。

# 按顺序开始 @KafkaListener s

一个常见的用例是，在另一个侦听器消耗了一个主题中的所有记录之后，启动一个侦听器。例如，在处理来自其他主题的记录之前，你可能希望将一个或多个压缩主题的内容加载到内存中。从版本 2.7.3 开始，引入了一个新的组件 ContainerGroupSequencer 。它使用 @KafkaListener``containerGroup 属性将容器分组，并在当前组中的所有容器都空闲时启动下一个组中的容器。

用一个例子最好地说明这一点。

@KafkaListener(id = "listen1", topics = "topic1", containerGroup = "g1", concurrency = "2")
public void listen1(String in) {
@KafkaListener(id = "listen2", topics = "topic2", containerGroup = "g1", concurrency = "2")
public void listen2(String in) {
@KafkaListener(id = "listen3", topics = "topic3", containerGroup = "g2", concurrency = "2")
public void listen3(String in) {
@KafkaListener(id = "listen4", topics = "topic4", containerGroup = "g2", concurrency = "2")
public void listen4(String in) {
@Bean
ContainerGroupSequencer sequencer(KafkaListenerEndpointRegistry registry) {
    return new ContainerGroupSequencer(registry, 5000, "g1", "g2");

在这里，我们在两组中有 4 个听众， g1 和 g2 。

在应用程序上下文初始化期间，Sequencer 将提供的组中所有容器的 autoStartup 属性设置为 false 。它还将任何容器（还没有设置）的 idleEventInterval 设置为提供的值（在本例中为 5000ms）。然后，当应用程序上下文启动序列器时，第一组中的容器将被启动。当 ListenerContainerIdleEvent s 被接收时，每个容器中的每个单独的子容器都被停止。当 ConcurrentMessageListenerContainer 中的所有子容器被停止时，父容器被停止。当一个组中的所有容器都被停止时，下一个组中的容器将被启动。一个组中的组或容器的数量没有限制。

默认情况下，最终组（ g2 以上）中的容器在空闲时不会停止。要修改该行为，请将序列器上的 stopLastGroupWhenIdle 设置为 true 。

作为旁白；以前，每个组中的容器都被添加到类型 Collection<MessageListenerContainer> 的 Bean 中，其 Bean 名称为 containerGroup 。现在不推荐这些集合，而支持类型 ContainerGroup 的 bean，其 Bean 名称是组名，后缀为 .group ；在上面的示例中，将有 2 个 bean g1.group 和 g2.group 。 Collection bean 将在未来的版本中被删除。

# 使用 KafkaTemplate 接收

本节介绍如何使用 KafkaTemplate 接收消息。

从版本 2.8 开始，模板有四个 receive() 方法：

ConsumerRecord<K, V> receive(String topic, int partition, long offset);
ConsumerRecord<K, V> receive(String topic, int partition, long offset, Duration pollTimeout);
ConsumerRecords<K, V> receive(Collection<TopicPartitionOffset> requested);
ConsumerRecords<K, V> receive(Collection<TopicPartitionOffset> requested, Duration pollTimeout);

如你所见，你需要知道需要检索的记录的分区和偏移量；为每个操作创建（并关闭）一个新的 Consumer 。

使用最后两个方法，可以单独检索每个记录，并将结果组装到 ConsumerRecords 对象中。在为请求创建 TopicPartitionOffset s 时，只支持正的绝对偏移量。

# 4.1.5.侦听器容器属性

# 4.1.6.应用程序事件

以下 Spring 应用程序事件由侦听器容器及其使用者发布：

• ConsumerStartingEvent -在使用者线程第一次启动时发布，然后开始轮询。

• ConsumerStartedEvent -在使用者即将开始轮询时发布。

• ConsumerFailedToStartEvent -如果在 consumerStartTimeout 容器属性内没有 ConsumerStartingEvent 发布，则发布。此事件可能表示配置的任务执行器没有足够的线程来支持它所使用的容器及其并发性。当出现此情况时，还会记录错误消息。

• ListenerContainerIdleEvent :在 idleInterval （如果配置）中没有收到消息时发布。

• ListenerContainerNoLongerIdleEvent :在先前发布 ListenerContainerIdleEvent 后，当记录被消费时发布。

• ListenerContainerPartitionIdleEvent :在 idlePartitionEventInterval 中没有从该分区接收到消息时发布（如果已配置）。

• ListenerContainerPartitionNoLongerIdleEvent :当从以前发布过 ListenerContainerPartitionIdleEvent 的分区中消费一条记录时发布。

• NonResponsiveConsumerEvent :当消费者似乎在 poll 方法中被阻止时发布。

• ConsumerPartitionPausedEvent :当一个分区暂停时，由每个使用者发布。

• ConsumerPartitionResumedEvent :当一个分区被恢复时，由每个使用者发布。

• ConsumerPausedEvent :当容器暂停时，由每个使用者发布。

• ConsumerResumedEvent :当容器恢复时，由每个使用者发布。

• max.poll.interval.ms :在停止之前由每个消费者发布。

• ConsumerStoppedEvent :在消费者关闭后发布。见 螺纹安全 。

• ContainerStoppedEvent :当所有消费者都停止使用时发布。

ListenerContainerIdleEvent 具有以下属性：

• source :发布事件的侦听器容器实例。

• container :侦听器容器或父侦听器容器，如果源容器是一个子容器。

• id :侦听器 ID（或容器 Bean 名称）。

• idleTime :事件发布时容器处于空闲状态的时间。

• topicPartitions :在事件生成时容器被分配的主题和分区。

• consumer :对 Kafka Consumer 对象的引用。例如，如果先前调用了消费者的 pause() 方法，那么当接收到事件时，它可以 resume() 。

• paused :容器当前是否暂停。有关更多信息，请参见 暂停和恢复监听器容器 。

ListenerContainerNoLongerIdleEvent 具有相同的属性，但 idleTime 和 paused 除外。

ListenerContainerPartitionIdleEvent 具有以下属性：

• source :发布事件的侦听器容器实例。

• container :侦听器容器或父侦听器容器，如果源容器是一个子容器。

• id :侦听器 ID（或容器 Bean 名称）。

• idleTime :事件发布时，分区消耗的时间是空闲的。

• topicPartition :触发事件的主题和分区。

• consumer :对 Kafka Consumer 对象的引用。例如，如果先前调用了消费者的 pause() 方法，那么当接收到事件时，它可以 resume() 。

• paused :是否为该消费者暂停了该分区的消费。有关更多信息，请参见 暂停和恢复监听器容器 。

ListenerContainerPartitionNoLongerIdleEvent 具有相同的属性，但 idleTime 和 paused 除外。

NonResponsiveConsumerEvent 具有以下属性：

• source :发布事件的侦听器容器实例。

• container :侦听器容器或父侦听器容器，如果源容器是一个子容器。

• id :侦听器 ID（或容器 Bean 名称）。

• timeSinceLastPoll :容器上次调用 poll() 之前的时间。

• topicPartitions :在事件生成时容器被分配的主题和分区。

• consumer :对 Kafka Consumer 对象的引用。例如，如果先前调用了消费者的 pause() 方法，则在接收到事件时可以 resume() 。

• paused :容器当前是否暂停。有关更多信息，请参见 暂停和恢复监听器容器 。

ConsumerPausedEvent 、 ConsumerResumedEvent 和 ConsumerStopping 事件具有以下属性：

• source :发布事件的侦听器容器实例。

• container :侦听器容器或父侦听器容器，如果源容器是一个子容器。

• partitions :涉及 TopicPartition 实例。

ConsumerPartitionPausedEvent ， ConsumerPartitionResumedEvent 事件具有以下属性：

• source :发布事件的侦听器容器实例。

• container :侦听器容器或父侦听器容器，如果源容器是一个子容器。

• partition :涉及 TopicPartition 实例。

ConsumerStartingEvent ， ConsumerStartingEvent ， ConsumerFailedToStartEvent ， ConsumerStoppedEvent 和 ContainerStoppedEvent 事件具有以下属性：

• source :发布事件的侦听器容器实例。

• container :侦听器容器或父侦听器容器，如果源容器是一个子容器。

所有容器（无论是子容器还是父容器）发布 ContainerStoppedEvent 。对于父容器，源属性和容器属性是相同的。

此外， ConsumerStoppedEvent 还具有以下附加属性：

• reason

  • NORMAL -消费者正常停止（容器已停止）。

  • ERROR -a java.lang.Error 被抛出。

  • FENCED -对事务生成器进行了保护，并且 stopContainerWhenFenced 容器属性是 true 。

  • AUTH -一个 AuthenticationException 或 AuthorizationException 被抛出，并且 authExceptionRetryInterval 未配置。

  • NO_OFFSET -对于一个分区没有偏移量，并且 auto.offset.reset 策略是 none 。

你可以使用此事件在出现以下情况后重新启动容器：

if (event.getReason.equals(Reason.FENCED)) {
    event.getSource(MessageListenerContainer.class).start();

# 检测空闲和无响应的消费者

尽管效率很高，但异步用户的一个问题是检测它们何时空闲。如果一段时间内没有消息到达，你可能需要采取一些措施。

你可以将侦听器容器配置为在一段时间后没有消息传递的情况下发布 ListenerContainerIdleEvent 。当容器处于空闲状态时，每 idleEventInterval 毫秒就会发布一个事件。

要配置此功能，请在容器上设置 idleEventInterval 。下面的示例展示了如何做到这一点：

@Bean
public KafkaMessageListenerContainer(ConsumerFactory<String, String> consumerFactory) {
    ContainerProperties containerProps = new ContainerProperties("topic1", "topic2");
    containerProps.setIdleEventInterval(60000L);
    KafkaMessageListenerContainer<String, String> container = new KafKaMessageListenerContainer<>(...);
    return container;

下面的示例展示了如何为 @KafkaListener 设置 idleEventInterval :

@Bean
public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
    factory.getContainerProperties().setIdleEventInterval(60000L);
    return factory;

在每种情况下，当容器处于空闲状态时，每分钟都会发布一次事件。

如果由于某种原因，使用者 poll() 方法没有退出，则不会接收到任何消息，也不能生成空闲事件（这是 kafka-clients 的早期版本在无法访问代理时的一个问题）。在这种情况下，如果轮询不在 3x 内返回 pollTimeout 属性，则容器将发布 NonResponsiveConsumerEvent 。默认情况下，该检查在每个容器中每 30 秒执行一次。在配置侦听器容器时，可以通过在 ContainerProperties 中设置 monitorInterval （默认 30 秒）和 noPollThreshold （默认 3.0）属性来修改此行为。 noPollThreshold 应该大于 1.0 ，以避免由于比赛条件而导致虚假事件。接收这样的事件可以让你停止容器，从而唤醒消费者，使其可以停止。

从版本 2.6.2 开始，如果容器已经发布了 ListenerContainerIdleEvent ，那么当随后接收到一条记录时，它将发布 ListenerContainerNoLongerIdleEvent 。

# 事件消费

你可以通过实现 ApplicationListener 来捕获这些事件——或者是一个普通的侦听器，或者是一个缩小到只接收这个特定事件的侦听器。还可以使用 Spring Framework4.2 中介绍的 @EventListener 。

下一个示例将 @KafkaListener 和 @EventListener 合并为一个类。你应该理解，应用程序侦听器获取所有容器的事件，因此，如果你想根据哪个容器空闲来采取特定的操作，可能需要检查侦听器 ID。你也可以为此目的使用 @EventListener``condition 。

有关事件属性的信息，请参见 应用程序事件 。

该事件通常发布在使用者线程上，因此与 Consumer 对象交互是安全的。

下面的示例同时使用 @KafkaListener 和 @EventListener :

public class Listener {
    @KafkaListener(id = "qux", topics = "annotated")
    public void listen4(@Payload String foo, Acknowledgment ack) {
    @EventListener(condition = "event.listenerId.startsWith('qux-')")
    public void eventHandler(ListenerContainerIdleEvent event) {

# 空闲时的当前位置

请注意，你可以通过在侦听器中实现 ConsumerSeekAware 来获得检测到空闲时的当前位置。见 onIdleContainer() in 寻求一种特定的抵消 。

# 4.1.7.主题/分区初始偏移

有几种方法可以设置分区的初始偏移量。

当手动分配分区时，可以在配置的 TopicPartitionOffset 参数中设置初始偏移量（如果需要）（参见 消息监听器容器 ）。你还可以在任何时候寻求特定的偏移。

在使用分组管理时，代理将分配分区：

• 对于新的 group.id ，初始偏移量由 auto.offset.reset 消费者属性（ earliest 或 latest ）确定。

• 对于现有的组 ID，初始偏移量是该组 ID 的当前偏移量。但是，你可以在初始化期间（或之后的任何时间）寻求特定的偏移量。

# 4.1.8.寻求一种特定的抵消

为了进行查找，侦听器必须实现 ConsumerSeekAware ，它具有以下方法：

void registerSeekCallback(ConsumerSeekCallback callback);
void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback);
void onPartitionsRevoked(Collection<TopicPartition> partitions)
void onIdleContainer(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback);

在启动容器时和分配分区时调用 registerSeekCallback 。在初始化后的某个任意时间进行查找时，应该使用此回调。你应该保存对回调的引用。如果在多个容器（或 ConcurrentMessageListenerContainer ）中使用相同的侦听器，则应将回调存储在 ThreadLocal 或由侦听器 Thread 键控的其他结构中。

当使用组管理时，分配分区时调用 onPartitionsAssigned 。例如，你可以使用这个方法，通过调用回调来设置分区的初始偏移量。你还可以使用此方法将此线程的回调与分配的分区关联起来（请参见下面的示例）。你必须使用回调参数，而不是传递到 registerSeekCallback 的参数。从版本 2.5.5 开始，即使使用 手动分区分配 ，也会调用此方法。

onPartitionsRevoked 在停止容器或 Kafka 撤销分配时调用。你应该放弃这个线程的回调，并删除与已撤销分区的任何关联。

回调有以下方法：

void seek(String topic, int partition, long offset);
void seekToBeginning(String topic, int partition);
void seekToBeginning(Collection=<TopicPartitions> partitions);
void seekToEnd(String topic, int partition);
void seekToEnd(Collection=<TopicPartitions> partitions);
void seekRelative(String topic, int partition, long offset, boolean toCurrent);
void seekToTimestamp(String topic, int partition, long timestamp);
void seekToTimestamp(Collection<TopicPartition> topicPartitions, long timestamp);

seekRelative 在版本 2.3 中被添加，以执行相对查找。

• offset 负且 toCurrent``false -相对于分区的末尾进行查找。

• offset 正和 toCurrent``false -相对于分区的开始进行查找。

• offset 负数和 toCurrent``true -相对于当前位置进行查找（倒带）。

• offset 正和 toCurrent``true -相对于当前位置进行搜索（快进）。

在版本 2.3 中还添加了 seekToTimestamp 方法。

当检测到空闲容器时，还可以从 onIdleContainer() 执行查找操作。有关如何启用空闲容器检测，请参见 检测空闲和无响应的消费者 。

public class MyListener implements ConsumerSeekAware {
    @Override
    public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
        callback.seekToBeginning(assignments.keySet());

要在运行时任意查找，请使用来自 registerSeekCallback 的回调引用来查找合适的线程。

下面是一个简单的 Spring 启动应用程序，它演示了如何使用回调；它向主题发送 10 条记录；在控制台中点击 <Enter> ，将导致所有分区从头开始查找。

@SpringBootApplication
public class SeekExampleApplication {
    public static void main(String[] args) {
        SpringApplication.run(SeekExampleApplication.class, args);
    @Bean
    public ApplicationRunner runner(Listener listener, KafkaTemplate<String, String> template) {
        return args -> {
            IntStream.range(0, 10).forEach(i -> template.send(
                new ProducerRecord<>("seekExample", i % 3, "foo", "bar")));
            while (true) {
                System.in.read();
                listener.seekToStart();
    @Bean
    public NewTopic topic() {
        return new NewTopic("seekExample", 3, (short) 1);
@Component
class Listener implements ConsumerSeekAware {
    private static final Logger logger = LoggerFactory.getLogger(Listener.class);
    private final ThreadLocal<ConsumerSeekCallback> callbackForThread = new ThreadLocal<>();
    private final Map<TopicPartition, ConsumerSeekCallback> callbacks = new ConcurrentHashMap<>();
    @Override
    public void registerSeekCallback(ConsumerSeekCallback callback) {
        this.callbackForThread.set(callback);
    @Override
    public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
        assignments.keySet().forEach(tp -> this.callbacks.put(tp, this.callbackForThread.get()));
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        partitions.forEach(tp -> this.callbacks.remove(tp));
        this.callbackForThread.remove();
    @Override
    public void onIdleContainer(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
    @KafkaListener(id = "seekExample", topics = "seekExample", concurrency = "3")
    public void listen(ConsumerRecord<String, String> in) {
        logger.info(in.toString());
    public void seekToStart() {
        this.callbacks.forEach((tp, callback) -> callback.seekToBeginning(tp.topic(), tp.partition()));

为了使事情变得更简单，版本 2.3 添加了 AbstractConsumerSeekAware 类，该类跟踪主题/分区将使用哪个回调。下面的示例展示了每次容器空闲时，如何查找每个分区中处理的最后一条记录。它还有一些方法，允许任意外部调用通过一条记录来倒带分区。

public class SeekToLastOnIdleListener extends AbstractConsumerSeekAware {
    @KafkaListener(id = "seekOnIdle", topics = "seekOnIdle")
    public void listen(String in) {
    @Override
    public void onIdleContainer(Map<org.apache.kafka.common.TopicPartition, Long> assignments,
            ConsumerSeekCallback callback) {
            assignments.keySet().forEach(tp -> callback.seekRelative(tp.topic(), tp.partition(), -1, true));
    * Rewind all partitions one record.
    public void rewindAllOneRecord() {
        getSeekCallbacks()
            .forEach((tp, callback) ->
                callback.seekRelative(tp.topic(), tp.partition(), -1, true));
    * Rewind one partition one record.
    public void rewindOnePartitionOneRecord(String topic, int partition) {
        getSeekCallbackFor(new org.apache.kafka.common.TopicPartition(topic, partition))
            .seekRelative(topic, partition, -1, true);

版本 2.6 为抽象类添加了方便的方法：

• seekToBeginning() -将所有分配的分区查找到开始位置

• seekToEnd() -将所有分配的分区查找到末尾

• seekToTimestamp(long time) -将所有分配的分区查找到由该时间戳表示的偏移量。

示例：

public class MyListener extends AbstractConsumerSeekAware {
    @KafkaListener(...)
    void listn(...) {
public class SomeOtherBean {
    MyListener listener;
    void someMethod() {
        this.listener.seekToTimestamp(System.currentTimeMillis - 60_000);

# 4.1.9.集装箱工厂

正如[ @KafkaListener 注释]（#kafka-listener-annotation）中所讨论的， ConcurrentKafkaListenerContainerFactory 用于为带注释的方法创建容器。

从版本 2.2 开始，你可以使用相同的工厂来创建任何 ConcurrentMessageListenerContainer 。如果你希望创建几个具有类似属性的容器，或者希望使用一些外部配置的工厂，例如 Spring Boot Auto-Configuration 提供的容器，那么这可能会很有用。创建容器后，你可以进一步修改其属性，其中许多属性是通过使用 container.getContainerProperties() 设置的。以下示例配置 ConcurrentMessageListenerContainer :

@Bean
public ConcurrentMessageListenerContainer<String, String>(
        ConcurrentKafkaListenerContainerFactory<String, String> factory) {
    ConcurrentMessageListenerContainer<String, String> container =
        factory.createContainer("topic1", "topic2");
    container.setMessageListener(m -> { ... } );
    return container;

从版本 2.3.4 开始，你可以向工厂添加 ContainerCustomizer ，以便在创建和配置每个容器之后进一步配置它。

@Bean
public KafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
    factory.setContainerCustomizer(container -> { /* customize the container */ });
    return factory;

# 4.1.10.螺纹安全

当使用并发消息侦听器容器时，将在所有使用者线程上调用单个侦听器实例。因此，侦听器需要是线程安全的，最好是使用无状态侦听器。如果不可能使你的侦听器线程安全，或者添加同步将大大降低添加并发性的好处，那么你可以使用以下几种技术中的一种：

• 使用 n 容器和 concurrency=1 原型作用域 MessageListener Bean，以便每个容器获得自己的实例（当使用 @KafkaListener 时，这是不可能的）。

• 将状态保持在 ThreadLocal<?> 实例中。

• 将 singleton 侦听器委托给在 SimpleThreadScope （或类似的作用域）中声明的 Bean。

为了便于清理线程状态（对于前面列表中的第二个和第三个项目），从版本 2.2 开始，侦听器容器在每个线程退出时发布 ConsumerStoppedEvent 。你可以使用 ApplicationListener 或 @EventListener 方法来使用这些事件，以从作用域中删除 ThreadLocal<?> 实例或 remove() 线程作用域 bean。请注意， SimpleThreadScope 不会销毁具有销毁接口的 bean（例如 DisposableBean ），因此你应该 destroy() 自己的实例。

# 4.1.11.监测

# 监视侦听器性能

从版本 2.3 开始，如果在类路径上检测到 Micrometer ，并且在应用程序上下文中存在一个 MeterRegistry ，则侦听器容器将自动为侦听器创建和更新微米计 Timer s。可以通过将 ContainerProperty``micrometerEnabled 设置为 false 来禁用计时器。

两个计时器被维护-一个用于对听者的成功调用，另一个用于失败调用。

计时器名为 spring.kafka.listener ，并具有以下标记：

• name :（容器 Bean 名称）

• result : success 或 failure

• exception : none 或 ListenerExecutionFailedException

你可以使用 ContainerProperties``micrometerTags 属性添加其他标记。

# 监控 Kafkatemplate 性能

从版本 2.5 开始，如果在类路径上检测到 Micrometer ，并且在应用程序上下文中存在一个 MeterRegistry ，则模板将自动为发送操作创建和更新 Micrometer Timer s。可以通过将模板的 micrometerEnabled 属性设置为 false 来禁用计时器。

两个计时器被维护-一个用于对听者的成功调用，另一个用于失败调用。

计时器名为 spring.kafka.template ，并具有以下标记：

• name :（模板 Bean 名称）

• result : success 或 failure

• exception : none 或失败的异常类名

你可以使用模板的 micrometerTags 属性添加其他标记。

# 千分尺本机度量

从版本 2.5 开始，该框架提供 工厂监听器 来管理微米计 KafkaClientMetrics 实例，无论何时创建和关闭生产者和消费者。

要启用此功能，只需将侦听器添加到你的生产者和消费者工厂：

@Bean
public ConsumerFactory<String, String> myConsumerFactory() {
    Map<String, Object> configs = consumerConfigs();
    DefaultKafkaConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(configs);
    cf.addListener(new MicrometerConsumerListener<String, String>(meterRegistry(),
            Collections.singletonList(new ImmutableTag("customTag", "customTagValue"))));
    return cf;
@Bean
public ProducerFactory<String, String> myProducerFactory() {
    Map<String, Object> configs = producerConfigs();
    configs.put(ProducerConfig.CLIENT_ID_CONFIG, "myClientId");
    DefaultKafkaProducerFactory<String, String> pf = new DefaultKafkaProducerFactory<>(configs);
    pf.addListener(new MicrometerProducerListener<String, String>(meterRegistry(),
            Collections.singletonList(new ImmutableTag("customTag", "customTagValue"))));
    return pf;

传递给侦听器的消费者/生产者 id 被添加到计价器的标记中，标记名 spring.id 。

获取一个 Kafka 度量的示例

double count = this.meterRegistry.get("kafka.producer.node.incoming.byte.total")
                .tag("customTag", "customTagValue")
                .tag("spring.id", "myProducerFactory.myClientId-1")
                .functionCounter()
                .count()

为 StreamsBuilderFactoryBean 提供了类似的侦听器-参见 Kafkastreams 测微仪支持 。

# 4.1.12.交易

本节描述了 Spring for Apache Kafka 如何支持事务。

# 概述

0.11.0.0 客户端库增加了对事务的支持。 Spring For Apache Kafka 通过以下方式增加了支持：

• KafkaTransactionManager :用于正常的 Spring 事务支持（ @Transactional ， TransactionTemplate 等）。

• 事务性 KafkaMessageListenerContainer

• 具有 KafkaTemplate 的本地事务

• 与其他事务管理器的事务同步

通过提供 DefaultKafkaProducerFactory 和 transactionIdPrefix 来启用事务。在这种情况下，工厂维护事务生产者的缓存，而不是管理单个共享的 Producer 。当用户在生成器上调用 close() 时，它将返回到缓存中进行重用，而不是实际关闭。每个生成器的 transactional.id 属性是 transactionIdPrefix + n ，其中 n 以 n 开头，并为每个新生成器递增，除非事务是由具有基于记录的侦听器的侦听器容器启动的。在这种情况下， transactional.id 是 <transactionIdPrefix>.<group.id>.<topic>.<partition> 。这是正确支持击剑僵尸， 如此处所述 (opens new window) 。在 1.3.7、2.0.6、2.1.10 和 2.2.0 版本中添加了这种新行为。如果希望恢复到以前的行为，可以将 DefaultKafkaProducerFactory 上的 producerPerConsumerPartition 属性设置为 false 。

另见[ transactionIdPrefix ]（#transaction-id-prefix）。

有了 Spring boot，只需要设置 spring.kafka.producer.transaction-id-prefix 属性-boot 将自动配置一个 KafkaTransactionManager Bean 并将其连接到侦听器容器中。

# 使用 KafkaTransactionManager

KafkaTransactionManager 是 Spring 框架 PlatformTransactionManager 的一个实现。为生产厂在其构造中的应用提供了参考.如果你提供了一个自定义的生产者工厂，那么它必须支持事务。见 ProducerFactory.transactionCapable() 。

你可以使用具有正常 Spring 事务支持的 KafkaTransactionManager （ @Transactional 、 TransactionTemplate 等）。如果事务是活动的，则在事务范围内执行的任何 KafkaTemplate 操作都使用事务的 Producer 。Manager 根据成功或失败提交或回滚事务。你必须配置 KafkaTemplate 以使用与事务管理器相同的 ProducerFactory 。

# 事务同步

本节引用仅生产者事务（不是由侦听器容器启动的事务）；有关在容器启动事务时链接事务的信息，请参见 使用消费者发起的交易 。

如果希望将记录发送到 Kafka 并执行某些数据库更新，则可以使用普通的 Spring 事务管理，例如，使用 DataSourceTransactionManager 。

@Transactional
public void process(List<Thing> things) {
    things.forEach(thing -> this.kafkaTemplate.send("topic", thing));
    updateDb(things);

@Transactional 注释的拦截器将启动事务， KafkaTemplate 将与该事务管理器同步一个事务；每个发送都将参与该事务。当该方法退出时，数据库事务将提交，然后是 Kafka 事务。如果希望以相反的顺序执行提交（首先是 Kafka），请使用嵌套的 @Transactional 方法，外部方法配置为使用 DataSourceTransactionManager ，内部方法配置为使用 KafkaTransactionManager 。

有关在 Kafka-first 或 DB-first 配置中同步 JDBC 和 Kafka 事务的应用程序示例，请参见[[ex-jdbc-sync]]。

# 使用消费者发起的事务

从版本 2.7 开始， ChainedKafkaTransactionManager 现在已被弃用；有关更多信息，请参见 Javadocs 的超类 ChainedTransactionManager 。相反，在容器中使用 KafkaTransactionManager 来启动 Kafka 事务，并用 @Transactional 注释侦听器方法来启动另一个事务。

有关链接 JDBC 和 Kafka 事务的示例应用程序，请参见[[ex-jdbc-sync]]。

# KafkaTemplate 本地事务

你可以使用 KafkaTemplate 在本地事务中执行一系列操作。下面的示例展示了如何做到这一点：

boolean result = template.executeInTransaction(t -> {
    t.sendDefault("thing1", "thing2");
    t.sendDefault("cat", "hat");
    return true;

回调中的参数是模板本身（ this ）。如果回调正常退出，则提交事务。如果抛出异常，事务将被回滚。

# transactionIdPrefix

正如 概述 中提到的，生产者工厂配置了此属性，以构建生产者 transactional.id 属性。在使用 EOSMode.ALPHA 运行应用程序的多个实例时，当在监听器容器线程上生成记录时，在所有实例上都必须相同，以满足 fencing zombies（在概述中也提到了）的要求。但是，当使用侦听器容器启动的 不是 事务生成记录时，每个实例的前缀必须不同。版本 2.3 使此配置更简单，尤其是在 Spring 启动应用程序中。在以前的版本中，你必须创建两个生产者工厂和 KafkaTemplate S-一个用于在侦听器容器线程上生成记录，另一个用于由 kafkaTemplate.executeInTransaction() 或由 @Transactional 方法上的事务拦截器启动的独立事务。

现在，你可以在 KafkaTemplate 和 KafkaTransactionManager 上覆盖工厂的 transactionalIdPrefix 。

当为侦听器容器使用事务管理器和模板时，通常将其默认设置为生产者工厂的属性。当使用 EOSMode.ALPHA 时，对于所有应用程序实例，该值应该是相同的。对于 EOSMode.BETA ，不再需要使用相同的 transactional.id ，即使对于消费者发起的事务也是如此；实际上，它必须在每个实例上都是唯一的，就像生产者发起的事务一样。对于由模板（或 @Transaction 的事务管理器）启动的事务，应该分别在模板和事务管理器上设置属性。此属性在每个应用程序实例上必须具有不同的值。

当使用 EOSMode.BETA （代理版本 >=2.5）时，此问题（ transactional.id 的不同规则）已被消除；请参见 一次语义学 。

# KafkaTemplate 事务性和非事务性发布

通常，当 KafkaTemplate 是事务性的（配置了能够处理事务的生产者工厂）时，事务是必需的。事务可以通过 TransactionTemplate 、 @Transactional 方法启动，调用 executeInTransaction ，或者在配置 KafkaTransactionManager 时通过侦听器容器启动。在事务范围之外使用模板的任何尝试都会导致模板抛出 IllegalStateException 。从版本 2.4.3 开始，你可以将模板的 allowNonTransactional 属性设置为 true 。在这种情况下，通过调用 ProducerFactory 的 createNonTransactionalProducer() 方法，模板将允许操作在没有事务的情况下运行；生产者将被缓存或线程绑定，以进行正常的重用。参见[使用 DefaultKafkaProducerFactory ]（#producer-factory）。

# 具有批处理侦听器的事务

当侦听器在使用事务时失败时，将调用 AfterRollbackProcessor 在回滚发生后采取一些操作。当在记录侦听器中使用默认的 AfterRollbackProcessor 时，将执行查找，以便重新交付失败的记录。但是，对于批处理侦听器，整个批处理将被重新交付，因为框架不知道批处理中的哪个记录失败了。有关更多信息，请参见 后回滚处理器 。

在使用批处理侦听器时，版本 2.4.2 引入了一种替代机制来处理批处理过程中的故障； BatchToRecordAdapter 。当将 batchListener 设置为 true 的容器工厂配置为 BatchToRecordAdapter 时，侦听器一次用一条记录调用。这允许在批处理中进行错误处理，同时仍然可以根据异常类型停止处理整个批处理。提供了一个默认的 BatchToRecordAdapter ，可以使用标准的 ConsumerRecordRecoverer 进行配置，例如 DeadLetterPublishingRecoverer 。下面的测试用例配置片段演示了如何使用此功能：

public static class TestListener {
    final List<String> values = new ArrayList<>();
    @KafkaListener(id = "batchRecordAdapter", topics = "test")
    public void listen(String data) {
        values.add(data);
        if ("bar".equals(data)) {
            throw new RuntimeException("reject partial");
@Configuration
@EnableKafka
public static class Config {
    ConsumerRecord<?, ?> failed;
    @Bean
    public TestListener test() {
        return new TestListener();
    @Bean
    public ConsumerFactory<?, ?> consumerFactory() {
        return mock(ConsumerFactory.class);
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
        factory.setConsumerFactory(consumerFactory());
        factory.setBatchListener(true);
        factory.setBatchToRecordAdapter(new DefaultBatchToRecordAdapter<>((record, ex) ->  {
            this.failed = record;
        return factory;

# 4.1.13.一次语义学

你可以为侦听器容器提供一个 KafkaAwareTransactionManager 实例。当这样配置时，容器在调用侦听器之前启动一个事务。侦听器执行的任何 KafkaTemplate 操作都参与事务。如果侦听器在使用 BatchMessageListener 时成功地处理该记录（或多个记录），则容器在事务管理器提交事务之前通过使用 producer.sendOffsetsToTransaction() 向事务发送偏移量。如果侦听器抛出异常，事务将被回滚，使用者将被重新定位，以便在下一次投票时可以检索回滚记录。有关更多信息和处理多次失败的记录，请参见 后回滚处理器 。

使用事务可以实现精确的一次语义（EOS）。

这意味着，对于 read→process-write 序列，可以保证 序列 恰好完成一次。（read 和 process 至少有一次语义）。

Spring 对于 Apache Kafka 版本 2.5 及更高版本，支持两种 EOS 模式：

• ALPHA - V1 的别名（不推荐）

• BETA - V2 的别名（不推荐）

• V1 -AKA transactional.id 击剑（自版本 0.11.0.0 起）

• V2 -AKAfetch-offset-request fencing（自版本 2.5 起）

在模式 V1 下，如果启动了另一个具有相同 transactional.id 的实例，那么生产者将被“隔离”。 Spring 通过对每个 group.id/topic/partition 使用 Producer 来管理这一点；当重新平衡发生时，新实例将使用相同的 transactional.id ，并且旧的生产者将被隔离。

对于模式 V2 ，不需要为每个 group.id/topic/partition 都有一个生产者，因为消费者元数据与偏移量一起发送到事务，并且代理可以确定生产者是否使用该信息来保护生产者。

从版本 2.6 开始，默认的 EOSMode 是 V2 。

要将容器配置为使用模式 ALPHA ，请将容器属性 EOSMode 设置为 ALPHA ，以恢复到以前的行为。

当你的代理程序升级到 2.5 或更高时，你应该将模式切换到 V2 ，但是生产者的数量将保持不变。然后可以对应用程序进行滚动升级，将 producerPerConsumerPartition 设置为 false ，以减少生成器的数量；还应该不再设置 subBatchPerPartition 容器属性。

如果你的代理已经是 2.5 或更新版本，则应该将 DefaultKafkaProducerFactory``producerPerConsumerPartition 属性设置为 false ，以减少所需的生产者数量。

当使用 V2 模式时，不再需要将 subBatchPerPartition 设置为 true ；当 EOSMode 为 V2 时，将默认为 false 。

有关更多信息，请参见 KIP-447 (opens new window) 。

V1 和 V2 以前是 ALPHA 和 BETA ；它们已被更改以使框架与 KIP-732 (opens new window) 对齐。

# 4.1.14.将 Spring bean 连接到生产者/消费者拦截器

Apache Kafka 提供了一种向生产者和消费者添加拦截器的机制。这些对象是由 Kafka 管理的，而不是 Spring，因此正常的 Spring 依赖注入不适用于在依赖的 Spring bean 中连接。但是，你可以使用拦截器 config() 方法手动连接这些依赖项。下面的 Spring 引导应用程序展示了如何通过覆盖 Boot 的默认工厂将一些依赖的 Bean 添加到配置属性中来实现这一点。

@SpringBootApplication
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    @Bean
    public ConsumerFactory<?, ?> kafkaConsumerFactory(SomeBean someBean) {
        Map<String, Object> consumerProperties = new HashMap<>();
        // consumerProperties.put(..., ...)
        // ...
        consumerProperties.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, MyConsumerInterceptor.class.getName());
        consumerProperties.put("some.bean", someBean);
        return new DefaultKafkaConsumerFactory<>(consumerProperties);
    @Bean
    public ProducerFactory<?, ?> kafkaProducerFactory(SomeBean someBean) {
        Map<String, Object> producerProperties = new HashMap<>();
        // producerProperties.put(..., ...)
        // ...
        Map<String, Object> producerProperties = properties.buildProducerProperties();
        producerProperties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, MyProducerInterceptor.class.getName());
        producerProperties.put("some.bean", someBean);
        DefaultKafkaProducerFactory<?, ?> factory = new DefaultKafkaProducerFactory<>(producerProperties);
        return factory;
    @Bean
    public SomeBean someBean() {
        return new SomeBean();
    @KafkaListener(id = "kgk897", topics = "kgh897")
    public void listen(String in) {
        System.out.println("Received " + in);
    @Bean
    public ApplicationRunner runner(KafkaTemplate<String, String> template) {
        return args -> template.send("kgh897", "test");
    @Bean
    public NewTopic kRequests() {
        return TopicBuilder.name("kgh897")
            .partitions(1)
            .replicas(1)
            .build();

public class SomeBean {
    public void someMethod(String what) {
        System.out.println(what + " in my foo bean");

public class MyProducerInterceptor implements ProducerInterceptor<String, String> {
    private SomeBean bean;
    @Override
    public void configure(Map<String, ?> configs) {
        this.bean = (SomeBean) configs.get("some.bean");
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        this.bean.someMethod("producer interceptor");
        return record;
    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
    @Override
    public void close() {

public class MyConsumerInterceptor implements ConsumerInterceptor<String, String> {
    private SomeBean bean;
    @Override
    public void configure(Map<String, ?> configs) {
        this.bean = (SomeBean) configs.get("some.bean");
    @Override
    public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> records) {
        this.bean.someMethod("consumer interceptor");
        return records;
    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
    @Override
    public void close() {

结果：

producer interceptor in my foo bean
consumer interceptor in my foo bean
Received test

# 4.1.15.暂停和恢复监听器容器

版本 2.1.3 为侦听器容器添加了 pause() 和 resume() 方法。以前，你可以在 ConsumerAwareMessageListener 中暂停一个消费者，并通过监听 ListenerContainerIdleEvent 来恢复它，该监听提供了对 Consumer 对象的访问。虽然可以通过使用事件侦听器在空闲容器中暂停使用者，但在某些情况下，这不是线程安全的，因为不能保证在使用者线程上调用事件侦听器。为了安全地暂停和恢复消费者，你应该在侦听器容器上使用 pause 和 resume 方法。a pause() 在下一个 poll() 之前生效；a resume() 在当前 poll() 返回之后生效。当容器暂停时，它将继续 poll() 使用者，从而避免在使用组管理时进行重新平衡，但它不会检索任何记录。有关更多信息，请参见 Kafka 文档。

从版本 2.1.5 开始，你可以调用 isPauseRequested() 来查看是否调用了 pause() 。但是，消费者可能还没有真正暂停。 isConsumerPaused() 如果所有 Consumer 实例都实际暂停，则返回 true。

此外（也是从 2.1.5 开始）， ConsumerPausedEvent 和 ConsumerResumedEvent 实例与容器一起作为 source 属性和 TopicPartition 属性所涉及的实例一起发布。

以下简单的 Spring 引导应用程序演示了如何使用容器注册中心获得对 @KafkaListener 方法的容器的引用，并暂停或恢复其使用者以及接收相应的事件：

@SpringBootApplication
public class Application implements ApplicationListener<KafkaEvent> {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args).close();
    @Override
    public void onApplicationEvent(KafkaEvent event) {
        System.out.println(event);
    @Bean
    public ApplicationRunner runner(KafkaListenerEndpointRegistry registry,
            KafkaTemplate<String, String> template) {
        return args -> {
            template.send("pause.resume.topic", "thing1");
            Thread.sleep(10_000);
            System.out.println("pausing");
            registry.getListenerContainer("pause.resume").pause();
            Thread.sleep(10_000);
            template.send("pause.resume.topic", "thing2");
            Thread.sleep(10_000);
            System.out.println("resuming");
            registry.getListenerContainer("pause.resume").resume();
            Thread.sleep(10_000);
    @KafkaListener(id = "pause.resume", topics = "pause.resume.topic")
    public void listen(String in) {
        System.out.println(in);
    @Bean
    public NewTopic topic() {
        return TopicBuilder.name("pause.resume.topic")
            .partitions(2)
            .replicas(1)
            .build();

下面的清单显示了前面示例的结果：

partitions assigned: [pause.resume.topic-1, pause.resume.topic-0]
thing1
pausing
ConsumerPausedEvent [partitions=[pause.resume.topic-1, pause.resume.topic-0]]
resuming
ConsumerResumedEvent [partitions=[pause.resume.topic-1, pause.resume.topic-0]]
thing2

# 4.1.16.在侦听器容器上暂停和恢复分区

从版本 2.7 开始，你可以通过使用侦听器容器中的 pausePartition(TopicPartition topicPartition) 和 resumePartition(TopicPartition topicPartition) 方法暂停并恢复分配给该使用者的特定分区的使用。暂停和恢复分别发生在 poll() 之前和之后，类似于 pause() 和 resume() 方法。如果请求了该分区的暂停， isPartitionPauseRequested() 方法将返回 true。如果该分区已有效地暂停， isPartitionPaused() 方法将返回 true。

另外，由于版本 2.7 ConsumerPartitionPausedEvent 和 ConsumerPartitionResumedEvent 实例与容器一起作为 source 属性和 TopicPartition 实例发布。

# 4.1.17.序列化、反序列化和消息转换

# 概述

Apache Kafka 提供了用于序列化和反序列化记录值及其键的高级 API。它存在于带有一些内置实现的 org.apache.kafka.common.serialization.Serializer<T> 和 org.apache.kafka.common.serialization.Deserializer<T> 抽象中。同时，我们可以通过使用 Producer 或 Consumer 配置属性来指定序列化器和反序列化器类。下面的示例展示了如何做到这一点：

props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

对于更复杂或更特殊的情况， KafkaConsumer （因此， KafkaProducer ）提供重载的构造函数来分别接受 Serializer 和 Deserializer 的实例。

当你使用这个 API 时， DefaultKafkaProducerFactory 和 DefaultKafkaConsumerFactory 还提供属性（通过构造函数或 setter 方法）来将自定义 Serializer 和 Deserializer 实例注入到目标 Producer 或 Consumer 中。同样，你可以通过构造函数传入 Supplier<Serializer> 或 Supplier<Deserializer> 实例-这些 Supplier s 在创建每个 Producer 或 Consumer 时被调用。

# 字符串序列化

自版本 2.5 以来， Spring for Apache Kafka 提供了 ToStringSerializer 和 ParseStringDeserializer 使用实体的字符串表示的类。它们依赖于方法 toString 和一些 Function<String> 或 BiFunction<String, Headers> 来解析字符串并填充实例的属性。通常，这会调用类上的一些静态方法，例如 parse :

ToStringSerializer<Thing> thingSerializer = new ToStringSerializer<>();
//...
ParseStringDeserializer<Thing> deserializer = new ParseStringDeserializer<>(Thing::parse);

默认情况下， ToStringSerializer 被配置为传递关于记录 Headers 中的序列化实体的类型信息。你可以通过将 addTypeInfo 属性设置为 false 来禁用它。此信息可由接收端的 ParseStringDeserializer 使用。

• ToStringSerializer.ADD_TYPE_INFO_HEADERS （默认 true ）：你可以将其设置为 false ，以在 ToStringSerializer 上禁用此功能（设置 addTypeInfo 属性）。

ParseStringDeserializer<Object> deserializer = new ParseStringDeserializer<>((str, headers) -> {
    byte[] header = headers.lastHeader(ToStringSerializer.VALUE_TYPE).value();
    String entityType = new String(header);
    if (entityType.contains("Thing")) {
        return Thing.parse(str);
    else {
        // ...parsing logic

可以配置用于将 String 转换为/from byte[] 的 Charset ，缺省值为 UTF-8 。

可以使用 ConsumerConfig 属性以解析器方法的名称配置反序列化器：

• ParseStringDeserializer.KEY_PARSER

• ParseStringDeserializer.VALUE_PARSER

属性必须包含类的完全限定名，后面跟着方法名，中间用一个句号 . 隔开。该方法必须是静态的，并且具有 (String, Headers) 或 (String) 的签名。

还提供了用于 Kafka 流的 ToFromStringSerde 。

# JSON

Spring 对于 Apache Kafka 还提供了基于 JacksonJSON 对象映射器的和实现。 JsonSerializer 允许将任何 Java 对象写为 JSON byte[] 。 JsonDeserializer 需要一个额外的 Class<?> targetType 参数，以允许将已使用的 byte[] 反序列化到正确的目标对象。下面的示例展示了如何创建 JsonDeserializer :

JsonDeserializer<Thing> thingDeserializer = new JsonDeserializer<>(Thing.class);

你可以使用 ObjectMapper 自定义 JsonSerializer 和 JsonDeserializer 。你还可以扩展它们，以在 configure(Map<String, ?> configs, boolean isKey) 方法中实现某些特定的配置逻辑。

从版本 2.3 开始，所有可感知 JSON 的组件都默认配置了 JacksonUtils.enhancedObjectMapper() 实例，该实例带有 MapperFeature.DEFAULT_VIEW_INCLUSION 和 DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES 禁用的功能。还为这样的实例提供了用于自定义数据类型的众所周知的模块，这样的 Java Time 和 Kotlin 支持。有关更多信息，请参见 JacksonUtils.enhancedObjectMapper() Javadocs。该方法还将 org.springframework.kafka.support.JacksonMimeTypeModule 对象序列化的 org.springframework.kafka.support.JacksonMimeTypeModule 注册到普通字符串中，以实现网络上的平台间兼容性。一个 JacksonMimeTypeModule 可以在应用程序上下文中注册为一个 Bean 并且它将被自动配置为[ Spring boot ObjectMapper 实例]（https://DOCS. Spring.io/ Spring-boot/DOCS/current/reference/html/howto- Spring-mvc.html#howto-customize-the-Jackson-objectmapper）。

同样从版本 2.3 开始， JsonDeserializer 提供了基于 TypeReference 的构造函数，以更好地处理目标泛型容器类型。

从版本 2.1 开始，你可以在记录 Headers 中传递类型信息，从而允许处理多个类型。此外，你可以通过使用以下 Kafka 属性来配置序列化器和反序列化器。如果分别为 KafkaConsumer 和 KafkaProducer 提供了 Deserializer 实例，则它们没有任何作用。

# 配置属性

• JsonSerializer.ADD_TYPE_INFO_HEADERS （默认 true ）：你可以将其设置为 false ，以在 JsonSerializer 上禁用此功能（设置 addTypeInfo 属性）。

• JsonSerializer.TYPE_MAPPINGS （默认 empty ）：见 映射类型 。

• JsonDeserializer.USE_TYPE_INFO_HEADERS （默认 true ）：可以将其设置为 false ，以忽略序列化器设置的头。

• JsonDeserializer.REMOVE_TYPE_INFO_HEADERS （默认 true ）：可以将其设置为 false ，以保留序列化器设置的标题。

• JsonDeserializer.KEY_DEFAULT_TYPE :如果不存在头信息，则用于对键进行反序列化的回退类型。

• JsonDeserializer.VALUE_DEFAULT_TYPE :如果不存在头信息，则用于反序列化值的回退类型。

• JsonDeserializer.TRUSTED_PACKAGES （默认 java.util ， java.lang ）：允许反序列化的以逗号分隔的包模式列表。 * 表示全部反序列化。

• JsonDeserializer.TYPE_MAPPINGS （默认 empty ）：见 映射类型 。

• JsonDeserializer.KEY_TYPE_METHOD （默认 empty ）：见 使用方法确定类型 。

• JsonDeserializer.VALUE_TYPE_METHOD （默认 empty ）：见 使用方法确定类型 。

从版本 2.2 开始，类型信息标头（如果由序列化器添加）将被反序列化器删除。可以通过将 removeTypeHeaders 属性设置为 false ，直接在反序列化器上或使用前面描述的配置属性，恢复到以前的行为。

另见[[tip-json]]。

# 映射类型

从版本 2.2 开始，当使用 JSON 时，你现在可以通过使用前面列表中的属性来提供类型映射。以前，你必须在序列化器和反序列化器中自定义类型映射器。映射由 token:className 对的逗号分隔列表组成。在出站时，有效负载的类名被映射到相应的令牌。在入站时，类型头中的令牌将映射到相应的类名。

下面的示例创建了一组映射：

senderProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
senderProps.put(JsonSerializer.TYPE_MAPPINGS, "cat:com.mycat.Cat, hat:com.myhat.hat");
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
consumerProps.put(JsonDeSerializer.TYPE_MAPPINGS, "cat:com.yourcat.Cat, hat:com.yourhat.hat");

如果使用 Spring Boot (opens new window) ，则可以在 application.properties （或 YAML）文件中提供这些属性。下面的示例展示了如何做到这一点：

spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
spring.kafka.producer.properties.spring.json.type.mapping=cat:com.mycat.Cat,hat:com.myhat.Hat

从版本 2.2 开始，你可以通过使用其中一个重载的构造函数，显式地将反序列化器配置为使用所提供的目标类型，并忽略头中的类型信息，该构造函数具有布尔值 useHeadersIfPresent （默认情况下是 true ）。下面的示例展示了如何做到这一点：

DefaultKafkaConsumerFactory<Integer, Cat1> cf = new DefaultKafkaConsumerFactory<>(props,
        new IntegerDeserializer(), new JsonDeserializer<>(Cat1.class, false));

# 使用方法确定类型

从版本 2.5 开始，你现在可以通过属性配置反序列化器来调用一个方法来确定目标类型。如果存在，这将覆盖上面讨论的任何其他技术。如果数据是由不使用 Spring 序列化器的应用程序发布的，并且你需要根据数据或其他头来反序列化到不同类型，那么这可能是有用的。将这些属性设置为方法名-一个完全限定的类名，后面跟着方法名，中间隔一个句号 . 。方法必须声明为 public static ，具有三个签名之一 (String topic, byte[] data, Headers headers) ， (byte[] data, Headers headers) 或 (byte[] data) ，并返回一个 Jackson JavaType 。

• JsonDeserializer.KEY_TYPE_METHOD : spring.json.key.type.method

• JsonDeserializer.VALUE_TYPE_METHOD : spring.json.value.type.method

你可以使用任意标题或检查数据来确定类型。

例子

JavaType thing1Type = TypeFactory.defaultInstance().constructType(Thing1.class);
JavaType thing2Type = TypeFactory.defaultInstance().constructType(Thing2.class);
public static JavaType thingOneOrThingTwo(byte[] data, Headers headers) {
    // {"thisIsAFieldInThing1":"value", ...
    if (data[21] == '1') {
        return thing1Type;
    else {
        return thing2Type;

对于更复杂的数据检查，可以考虑使用 JsonPath 或类似的方法，但是，确定类型的测试越简单，过程就会越有效。

以下是以编程方式（在构造函数中向消费者工厂提供反序列化器时）创建反序列化器的示例：

JsonDeserializer<Object> deser = new JsonDeserializer<>()
        .trustedPackages("*")
        .typeResolver(SomeClass::thing1Thing2JavaTypeForTopic);
public static JavaType thing1Thing2JavaTypeForTopic(String topic, byte[] data, Headers headers) {

# 纲领性建设

从版本 2.3 开始，当以编程方式构建在生产者/消费者工厂中使用的序列化器/反序列化器时，你可以使用 Fluent API，这简化了配置。

@Bean
public ProducerFactory<MyKeyType, MyValueType> pf() {
    Map<String, Object> props = new HashMap<>();
    // props.put(..., ...)
    // ...
    DefaultKafkaProducerFactory<MyKeyType, MyValueType> pf = new DefaultKafkaProducerFactory<>(props,
        new JsonSerializer<MyKeyType>()
            .forKeys()
            .noTypeInfo(),
        new JsonSerializer<MyValueType>()
            .noTypeInfo());
    return pf;
@Bean
public ConsumerFactory<MyKeyType, MyValueType> cf() {
    Map<String, Object> props = new HashMap<>();
    // props.put(..., ...)
    // ...
    DefaultKafkaConsumerFactory<MyKeyType, MyValueType> cf = new DefaultKafkaConsumerFactory<>(props,
        new JsonDeserializer<>(MyKeyType.class)
            .forKeys()
            .ignoreTypeHeaders(),
        new JsonDeserializer<>(MyValueType.class)
            .ignoreTypeHeaders());
    return cf;

要以编程方式提供类型映射，类似于 使用方法确定类型 ，请使用 typeFunction 属性。

例子

JsonDeserializer<Object> deser = new JsonDeserializer<>()
        .trustedPackages("*")
        .typeFunction(MyUtils::thingOneOrThingTwo);

或者，只要不使用 Fluent API 配置属性，或者不使用 set*() 方法设置属性，工厂将使用配置属性配置序列化器/反序列化器；参见 配置属性 。

# 委托序列化器和反序列化器

# 使用头文件

版本 2.3 引入了 DelegatingSerializer 和 DelegatingDeserializer ，它们允许使用不同的键和/或值类型来生成和消费记录。制作者必须将标题 DelegatingSerializer.VALUE_SERIALIZATION_SELECTOR 设置为选择器值，用于选择要使用哪个序列化器作为该值，而 DelegatingSerializer.KEY_SERIALIZATION_SELECTOR 作为该键；如果找不到匹配项，则抛出 IllegalStateException 。

对于传入的记录，反序列化器使用相同的头来选择要使用的反序列化器；如果未找到匹配项或头不存在，则返回 RAW byte[] 。

你可以通过构造函数将选择器的映射配置为 Serializer / Deserializer ，也可以通过 Kafka Producer/Consumer 属性配置它，使用键 DelegatingSerializer.VALUE_SERIALIZATION_SELECTOR_CONFIG 和 DelegatingSerializer.KEY_SERIALIZATION_SELECTOR_CONFIG 。对于序列化器，producer 属性可以是 Map<String, Object> ，其中键是选择器，值是 Serializer 实例，序列化器 Class 或类名。该属性也可以是一串以逗号分隔的映射项，如下所示。

对于反序列化器，消费者属性可以是 Map<String, Object> ，其中键是选择器，值是 Deserializer 实例，反序列化器 Class 或类名。该属性也可以是一串以逗号分隔的映射项，如下所示。

要配置使用属性，请使用以下语法：

producerProps.put(DelegatingSerializer.VALUE_SERIALIZATION_SELECTOR_CONFIG,
    "thing1:com.example.MyThing1Serializer, thing2:com.example.MyThing2Serializer")
consumerProps.put(DelegatingDeserializer.VALUE_SERIALIZATION_SELECTOR_CONFIG,
    "thing1:com.example.MyThing1Deserializer, thing2:com.example.MyThing2Deserializer")

然后，制作人将 DelegatingSerializer.VALUE_SERIALIZATION_SELECTOR 的标题设置为 thing1 或 thing2 。

这种技术支持向相同的主题（或不同的主题）发送不同的类型。

有关将不同类型发送到不同主题的另一种技术，请参见[使用 RoutingKafkaTemplate ]（#routing-template）。

# 按类型分列

2.8 版引入了 DelegatingByTypeSerializer 。

@Bean
public ProducerFactory<Integer, Object> producerFactory(Map<String, Object> config) {
    return new DefaultKafkaProducerFactory<>(config,
            null, new DelegatingByTypeSerializer(Map.of(
                    byte[].class, new ByteArraySerializer(),
                    Bytes.class, new BytesSerializer(),
                    String.class, new StringSerializer())));

从版本 2.8.3 开始，你可以将序列化器配置为检查是否可以从目标对象分配映射键，这在委托序列化器可以序列化子类时很有用。在这种情况下，如果有可亲的匹配，则应该提供一个有序的 Map ，例如一个 LinkedHashMap 。

# 按主题

从版本 2.8 开始， DelegatingByTopicSerializer 和 DelegatingByTopicDeserializer 允许基于主题名称选择序列化器/反序列化器。regex Pattern s 用于查找要使用的实例。可以使用构造函数或通过属性（用逗号分隔的列表 pattern:serializer ）来配置映射。

producerConfigs.put(DelegatingByTopicSerializer.VALUE_SERIALIZATION_TOPIC_CONFIG,
            "topic[0-4]:" + ByteArraySerializer.class.getName()
        + ", topic[5-9]:" + StringSerializer.class.getName());
ConsumerConfigs.put(DelegatingByTopicDeserializer.VALUE_SERIALIZATION_TOPIC_CONFIG,
            "topic[0-4]:" + ByteArrayDeserializer.class.getName()
        + ", topic[5-9]:" + StringDeserializer.class.getName());

使用 KEY_SERIALIZATION_TOPIC_CONFIG 作为键。

@Bean
public ProducerFactory<Integer, Object> producerFactory(Map<String, Object> config) {
    return new DefaultKafkaProducerFactory<>(config,
            null,
            new DelegatingByTopicSerializer(Map.of(
                    Pattern.compile("topic[0-4]"), new ByteArraySerializer(),
                    Pattern.compile("topic[5-9]"), new StringSerializer())),
                    new JsonSerializer<Object>());  // default

你可以使用 DelegatingByTopicSerialization.KEY_SERIALIZATION_TOPIC_DEFAULT 和 DelegatingByTopicSerialization.VALUE_SERIALIZATION_TOPIC_DEFAULT 指定一个默认的序列化器/反序列化器，当没有模式匹配时使用。

当设置为 false 时，另一个属性 DelegatingByTopicSerialization.CASE_SENSITIVE （默认 true ）会使主题查找不区分大小写。

# 重试反序列化器

RetryingDeserializer 使用委托 Deserializer 和 RetryTemplate 来重试反序列化，当委托在反序列化过程中可能出现瞬时错误时，例如网络问题。

ConsumerFactory cf = new DefaultKafkaConsumerFactory(myConsumerConfigs,
    new RetryingDeserializer(myUnreliableKeyDeserializer, retryTemplate),
    new RetryingDeserializer(myUnreliableValueDeserializer, retryTemplate));

请参阅 spring-retry (opens new window) 项目，以配置带有重试策略、Back off 策略等的 RetryTemplate 项目。

# Spring 消息传递消息转换

虽然 Serializer 和 Deserializer API 从低级别的 Kafka Consumer 和 Producer 透视图来看是非常简单和灵活的，但是在 Spring 消息传递级别，当使用 @KafkaListener 或 Spring Integration’s Apache Kafka Support (opens new window) 时，你可能需要更多的灵活性。为了让你能够轻松地转换 org.springframework.messaging.Message ， Spring for Apache Kafka 提供了一个 MessageConverter 的抽象，带有 MessagingMessageConverter 实现及其 JsonMessageConverter （和子类）定制。你可以直接将 MessageConverter 注入 KafkaTemplate 实例中，并使用 AbstractKafkaListenerContainerFactory Bean 对 @KafkaListener.containerFactory() 属性的定义。下面的示例展示了如何做到这一点：

@Bean
public KafkaListenerContainerFactory<?, ?> kafkaJsonListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setMessageConverter(new JsonMessageConverter());
    return factory;
@KafkaListener(topics = "jsonData",
                containerFactory = "kafkaJsonListenerContainerFactory")
public void jsonListener(Cat cat) {

在使用 Spring 引导时，只需将转换器定义为 @Bean ，并且 Spring 引导自动配置将其连接到自动配置的模板和容器工厂。

当使用 @KafkaListener 时，将向消息转换器提供参数类型，以帮助进行转换。

从版本 2.7.1 开始，可以将消息有效负载转换委托给 spring-messaging``SmartMessageConverter ；例如，这允许基于 MessageHeaders.CONTENT_TYPE 头进行转换。

当在 KafkaTemplate 和侦听器容器工厂中使用默认转换器时，你可以通过在模板上调用 setMessagingConverter() 和在 @KafkaListener 方法上通过 contentMessageConverter 属性来配置 SmartMessageConverter 。

例子：

template.setMessagingConverter(mySmartConverter);

@KafkaListener(id = "withSmartConverter", topics = "someTopic",
    contentTypeConverter = "mySmartConverter")
public void smart(Thing thing) {

# 使用 Spring 数据投影接口

从版本 2.1.1 开始，你可以将 JSON 转换为 Spring 数据投影接口，而不是具体的类型。这允许对数据进行非常有选择性的、低耦合的绑定，包括从 JSON 文档中的多个位置查找值。例如，以下接口可以定义为消息有效负载类型：

interface SomeSample {
  @JsonPath({ "$.username", "$.user.name" })
  String getUsername();

@KafkaListener(id="projection.listener", topics = "projection")
public void projection(SomeSample in) {
    String username = in.getUsername();

默认情况下，访问器方法将用于在接收的 JSON 文档中查找属性名称 AS 字段。 @JsonPath 表达式允许定制值查找，甚至可以定义多个 JSON 路径表达式，从多个位置查找值，直到表达式返回实际值。

要启用此功能，请使用配置有适当委托转换器的 ProjectingMessageConverter （用于出站转换和转换非投影接口）。你还必须将 spring-data:spring-data-commons 和 com.jayway.jsonpath:json-path 添加到类路径。

当用作 @KafkaListener 方法的参数时，接口类型将作为正常类型自动传递给转换器。

# 使用 ErrorHandlingDeserializer

当反序列化器无法对消息进行反序列化时， Spring 无法处理该问题，因为它发生在 poll() 返回之前。为了解决这个问题，引入了 ErrorHandlingDeserializer 。这个反序列化器委托给一个真正的反序列化器（键或值）。如果委托未能反序列化记录内容，则 ErrorHandlingDeserializer 在包含原因和原始字节的头文件中返回一个 null 值和一个 DeserializationException 值。当你使用一个记录级别 MessageListener 时，如果 ConsumerRecord 包含一个用于键或值的 DeserializationException 头，则使用失败的 ErrorHandler 调用容器的 ConsumerRecord 。记录不会传递给监听器。

或者，你可以通过提供 failedDeserializationFunction 来配置 ErrorHandlingDeserializer 以创建自定义值，这是 Function<FailedDeserializationInfo, T> 。调用此函数以创建 T 的实例，该实例将以通常的方式传递给侦听器。一个类型为 FailedDeserializationInfo 的对象，它包含提供给函数的所有上下文信息。你可以在头文件中找到 DeserializationException （作为序列化的 Java 对象）。有关更多信息，请参见 Javadoc (opens new window) 中的 ErrorHandlingDeserializer 。

你可以使用 DefaultKafkaConsumerFactory 构造函数，它接受键和值 Deserializer 对象，并在适当的 ErrorHandlingDeserializer 实例中连线，你已经用适当的委托进行了配置。或者，你可以使用消费者配置属性（ ErrorHandlingDeserializer 使用的属性）来实例化委托。属性名为 ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS 和 ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS 。属性值可以是类或类名。下面的示例展示了如何设置这些属性：

... // other props
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
props.put(ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS, JsonDeserializer.class);
props.put(JsonDeserializer.KEY_DEFAULT_TYPE, "com.example.MyKey")
props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class.getName());
props.put(JsonDeserializer.VALUE_DEFAULT_TYPE, "com.example.MyValue")
props.put(JsonDeserializer.TRUSTED_PACKAGES, "com.example")
return new DefaultKafkaConsumerFactory<>(props);

下面的示例使用 failedDeserializationFunction 。

public class BadFoo extends Foo {
  private final FailedDeserializationInfo failedDeserializationInfo;
  public BadFoo(FailedDeserializationInfo failedDeserializationInfo) {
    this.failedDeserializationInfo = failedDeserializationInfo;
  public FailedDeserializationInfo getFailedDeserializationInfo() {
    return this.failedDeserializationInfo;
public class FailedFooProvider implements Function<FailedDeserializationInfo, Foo> {
  @Override
  public Foo apply(FailedDeserializationInfo info) {
    return new BadFoo(info);

前面的示例使用以下配置：

...
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
consumerProps.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class);
consumerProps.put(ErrorHandlingDeserializer.VALUE_FUNCTION, FailedFooProvider.class);

@Bean
public ProducerFactory<String, Object> producerFactory() {
  return new DefaultKafkaProducerFactory<>(producerConfiguration(), new StringSerializer(),
    new DelegatingByTypeSerializer(Map.of(byte[].class, new ByteArraySerializer(),
          MyNormalObject.class, new JsonSerializer<Object>())));
@Bean
public KafkaTemplate<String, Object> kafkaTemplate() {
  return new KafkaTemplate<>(producerFactory());

当在批处理侦听器中使用 ErrorHandlingDeserializer 时，必须检查消息头中的反序列化异常。当与 DefaultBatchErrorHandler 一起使用时，你可以使用该头确定异常在哪个记录上失败，并通过 BatchListenerFailedException 与错误处理程序通信。

@KafkaListener(id = "test", topics = "test")
void listen(List<Thing> in, @Header(KafkaHeaders.BATCH_CONVERTED_HEADERS) List<Map<String, Object>> headers) {
    for (int i = 0; i < in.size(); i++) {
        Thing thing = in.get(i);
        if (thing == null
                && headers.get(i).get(SerializationUtils.VALUE_DESERIALIZER_EXCEPTION_HEADER) != null) {
            DeserializationException deserEx = ListenerUtils.byteArrayToDeserializationException(this.logger,
                    (byte[]) headers.get(i).get(SerializationUtils.VALUE_DESERIALIZER_EXCEPTION_HEADER));
            if (deserEx != null) {
                logger.error(deserEx, "Record at index " + i + " could not be deserialized");
            throw new BatchListenerFailedException("Deserialization", deserEx, i);
        process(thing);

ListenerUtils.byteArrayToDeserializationException() 可用于将标题转换为 DeserializationException 。

在消费 List<ConsumerRecord<?, ?> 时，使用 ListenerUtils.getExceptionFromHeader() 代替：

@KafkaListener(id = "kgh2036", topics = "kgh2036")
void listen(List<ConsumerRecord<String, Thing>> in) {
    for (int i = 0; i < in.size(); i++) {
        ConsumerRecord<String, Thing> rec = in.get(i);
        if (rec.value() == null) {
            DeserializationException deserEx = ListenerUtils.getExceptionFromHeader(rec,
                    SerializationUtils.VALUE_DESERIALIZER_EXCEPTION_HEADER, this.logger);
            if (deserEx != null) {
                logger.error(deserEx, "Record at offset " + rec.offset() + " could not be deserialized");
                throw new BatchListenerFailedException("Deserialization", deserEx, i);
        process(rec.value());

# 与批处理侦听器的有效负载转换

在使用批监听器容器工厂时，还可以在 BatchMessagingMessageConverter 中使用 JsonMessageConverter 来转换批处理消息。有关更多信息，请参见 序列化、反序列化和消息转换 和 Spring Messaging Message Conversion 。

默认情况下，转换的类型是从侦听器参数推断出来的。如果将 JsonMessageConverter 配置为 DefaultJackson2TypeMapper ，并将其 TypePrecedence 设置为 TYPE_ID （而不是默认的 INFERRED ），则转换器将使用头中的类型信息（如果存在的话）。例如，这允许使用接口声明侦听器方法，而不是使用具体的类。此外，类型转换器支持映射，因此反序列化可以是与源不同的类型（只要数据是兼容的）。当你使用[class-level @KafkaListener 实例]（#class-level-kafkalistener）时，这也很有用，因为有效负载必须已经被转换，以确定要调用的方法。下面的示例创建使用此方法的 bean:

@Bean
public KafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setBatchListener(true);
    factory.setMessageConverter(new BatchMessagingMessageConverter(converter()));
    return factory;
@Bean
public JsonMessageConverter converter() {
    return new JsonMessageConverter();

请注意，要使其工作，转换目标的方法签名必须是具有单一泛型参数类型的容器对象，例如：

@KafkaListener(topics = "blc1")
public void listen(List<Foo> foos, @Header(KafkaHeaders.OFFSET) List<Long> offsets) {

请注意，你仍然可以访问批处理头。

如果批处理转换器有一个支持它的记录转换器，那么你还可以接收一个消息列表，其中根据通用类型转换了有效负载。下面的示例展示了如何做到这一点：

@KafkaListener(topics = "blc3", groupId = "blc3")
public void listen1(List<Message<Foo>> fooMessages) {

# ConversionService 定制

从版本 2.1.1 开始，默认 org.springframework.core.convert.ConversionService 用于解析侦听器方法调用的参数所使用的 org.springframework.core.convert.ConversionService 与实现以下任何接口的所有 bean 一起提供：

• org.springframework.core.convert.converter.Converter

• org.springframework.core.convert.converter.GenericConverter

• org.springframework.format.Formatter

这使你可以进一步定制侦听器反序列化，而无需更改 ConsumerFactory 和 KafkaListenerContainerFactory 的默认配置。

# 将自定义 HandlerMethodArgumentResolver 添加到 @KafkaListener

从版本 2.4.2 开始，你可以添加自己的 HandlerMethodArgumentResolver 并解析自定义方法参数。你所需要的只是实现 KafkaListenerConfigurer 并使用来自类 setCustomMethodArgumentResolvers() 的方法 setCustomMethodArgumentResolvers() 。

@Configuration
class CustomKafkaConfig implements KafkaListenerConfigurer {
    @Override
    public void configureKafkaListeners(KafkaListenerEndpointRegistrar registrar) {
        registrar.setCustomMethodArgumentResolvers(
            new HandlerMethodArgumentResolver() {
                @Override
                public boolean supportsParameter(MethodParameter parameter) {
                    return CustomMethodArgument.class.isAssignableFrom(parameter.getParameterType());
                @Override
                public Object resolveArgument(MethodParameter parameter, Message<?> message) {
                    return new CustomMethodArgument(
                        message.getHeaders().get(KafkaHeaders.RECEIVED_TOPIC, String.class)

还可以通过在 KafkaListenerEndpointRegistrar Bean 中添加自定义 MessageHandlerMethodFactory 来完全替换框架的参数解析。如果你这样做，并且你的应用程序需要处理 Tombstone 记录，使用 null``value() （例如来自压缩主题），你应该向工厂添加 KafkaNullAwarePayloadArgumentResolver ；它必须是最后一个解析器，因为它支持所有类型，并且可以在没有 @Payload 注释的情况下匹配参数。如果你使用的是 DefaultMessageHandlerMethodFactory ，请将此解析器设置为最后一个自定义解析器；工厂将确保此解析器将在标准 PayloadMethodArgumentResolver 之前使用，该标准不知道 KafkaNull 的有效负载。

另见 “墓碑”记录的空载和日志压缩 。

# 4.1.18.消息头

0.11.0.0 客户机引入了对消息中的头的支持。从版本 2.0 开始， Spring for Apache Kafka 现在支持将这些头映射到 spring-messaging``MessageHeaders 。

Apache Kafka 头具有一个简单的 API，如下面的接口定义所示：

public interface Header {
    String key();
    byte[] value();

提供 KafkaHeaderMapper 策略来映射 Kafka Headers 和 MessageHeaders 之间的头条目。其接口定义如下：

public interface KafkaHeaderMapper {
    void fromHeaders(MessageHeaders headers, Headers target);
    void toHeaders(Headers source, Map<String, Object> target);

DefaultKafkaHeaderMapper 将键映射到 MessageHeaders 头名称，并且为了支持出站消息的丰富头类型，执行了 JSON 转换。一个“特殊”头（键为 spring_json_header_types ）包含一个 <key>:<type> 的 JSON 映射。这个头用于入站侧，以提供每个头值到原始类型的适当转换。

在入站方面，所有 Kafka Header 实例都映射到 MessageHeaders 。在出站端，默认情况下，所有 MessageHeaders 都被映射，除了 id ， timestamp ，以及映射到 ConsumerRecord 属性的头。

通过向映射器提供模式，你可以指定要为出站消息映射哪些头。下面的清单显示了一些示例映射：

public DefaultKafkaHeaderMapper() { (1)
public DefaultKafkaHeaderMapper(ObjectMapper objectMapper) { (2)
public DefaultKafkaHeaderMapper(String... patterns) { (3)
public DefaultKafkaHeaderMapper(ObjectMapper objectMapper, String... patterns) { (4)

模式非常简单，可以包含一个引导通配符（ **), a trailing wildcard, or both (for example, ** .cat.* ）。你可以使用一个前导 ! 来否定模式。与标头名称（无论是正的还是负的）相匹配的第一个模式获胜。

当你提供自己的模式时，我们建议包括 !id 和 !timestamp ，因为这些头在入站侧是只读的。

从版本 2.2.5 开始，你可以指定某些字符串值的头不应该使用 JSON 进行映射，而应该从 RAW byte[] 映射到/。 AbstractKafkaHeaderMapper 具有新的属性； mapAllStringsOut 当设置为 true 时，所有字符串值头将使用 byte[] 属性（默认 UTF-8 ）转换为 byte[] 。此外，还有一个属性 rawMappedHeaders ，它是 header name : boolean 的映射；如果映射包含一个头名称，并且头包含一个 String 值，则将使用字符集将其映射为一个 RAW byte[] 。此映射还用于使用字符集将原始传入的 byte[] 头映射到 String ，当且仅当映射值中的布尔值 true 。如果布尔值是 false ，或者标头名不在映射中，并且具有 true 值，则传入的标头会简单地映射为未映射的原始标头。

下面的测试用例演示了这种机制。

@Test
public void testSpecificStringConvert() {
    DefaultKafkaHeaderMapper mapper = new DefaultKafkaHeaderMapper();
    Map<String, Boolean> rawMappedHeaders = new HashMap<>();
    rawMappedHeaders.put("thisOnesAString", true);
    rawMappedHeaders.put("thisOnesBytes", false);
    mapper.setRawMappedHeaders(rawMappedHeaders);
    Map<String, Object> headersMap = new HashMap<>();
    headersMap.put("thisOnesAString", "thing1");
    headersMap.put("thisOnesBytes", "thing2");
    headersMap.put("alwaysRaw", "thing3".getBytes());
    MessageHeaders headers = new MessageHeaders(headersMap);
    Headers target = new RecordHeaders();
    mapper.fromHeaders(headers, target);
    assertThat(target).containsExactlyInAnyOrder(
            new RecordHeader("thisOnesAString", "thing1".getBytes()),
            new RecordHeader("thisOnesBytes", "thing2".getBytes()),
            new RecordHeader("alwaysRaw", "thing3".getBytes()));
    headersMap.clear();
    mapper.toHeaders(target, headersMap);
    assertThat(headersMap).contains(
            entry("thisOnesAString", "thing1"),
            entry("thisOnesBytes", "thing2".getBytes()),
            entry("alwaysRaw", "thing3".getBytes()));

默认情况下， DefaultKafkaHeaderMapper 在 MessagingMessageConverter 和 BatchMessagingMessageConverter 中使用 DefaultKafkaHeaderMapper ，只要 Jackson 在类路径上。

有了批处理转换器，转换后的头在 KafkaHeaders.BATCH_CONVERTED_HEADERS 中是可用的，如 List<Map<String, Object>> ，其中映射在列表的一个位置对应于数据在有效载荷中的位置。

如果没有转换器（要么是因为 Jackson 不存在，要么是显式地将其设置为 null ），则消费者记录的头在 KafkaHeaders.NATIVE_HEADERS 头中提供未转换的头。这个报头是 Headers 对象（或者在批处理转换器的情况下是 List<Headers> 对象），其中列表中的位置对应于有效负载中的数据位置）。

@Bean
MessagingMessageConverter converter() {
    MessagingMessageConverter converter = new MessagingMessageConverter();
    DefaultKafkaHeaderMapper mapper = new DefaultKafkaHeaderMapper();
    mapper.setEncodeStrings(true);
    converter.setHeaderMapper(mapper);
    return converter;

如果使用 Spring 引导，它将自动配置这个转换器 Bean 到自动配置的 KafkaTemplate 中；否则你应该将这个转换器添加到模板中。

# 4.1.19.“墓碑”记录的空载和日志压缩

当你使用 对数压缩 (opens new window) 时，你可以发送和接收带有 null 有效负载的消息，以识别删除的密钥。

由于其他原因，你也可以接收 null 值，例如，当不能反序列化某个值时，可能会返回 null 值。

要通过使用 KafkaTemplate 发送 null 有效负载，可以将 null 传递到 send() 方法的值参数中。这方面的一个例外是 send(Message<?> message) 变体。由于 spring-messaging``Message<?> 不能具有 null 有效载荷，因此可以使用一种称为 KafkaNull 的特殊有效载荷类型，并且框架发送 null 。为了方便起见，提供了静态 KafkaNull.INSTANCE 。

当使用消息侦听器容器时，接收到的 ConsumerRecord 具有 null``value() 。

要将 @KafkaListener 配置为处理 null 有效负载，必须使用 @Payload 注释和 required = false 。如果这是一个压缩日志的墓碑消息，那么你通常还需要这个键，这样你的应用程序就可以确定哪个键被“删除”了。下面的示例展示了这样的配置：

@KafkaListener(id = "deletableListener", topics = "myTopic")
public void listen(@Payload(required = false) String value, @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String key) {
    // value == null represents key deletion

当使用具有多个 @KafkaHandler 方法的类级 @KafkaListener 时，需要进行一些额外的配置。具体地说，你需要一个带有 @KafkaHandler 有效负载的 KafkaNull 方法。下面的示例展示了如何配置一个：

@KafkaListener(id = "multi", topics = "myTopic")
static class MultiListenerBean {
    @KafkaHandler
    public void listen(String cat) {
    @KafkaHandler
    public void listen(Integer hat) {
    @KafkaHandler
    public void delete(@Payload(required = false) KafkaNull nul, @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) int key) {

请注意，参数是 null ，而不是 KafkaNull 。

# 4.1.20.处理异常

本节描述了如何处理在使用 Spring 用于 Apache Kafka 时可能出现的各种异常。

# 侦听器错误处理程序

从版本 2.0 开始， @KafkaListener 注释有一个新属性： errorHandler 。

你可以使用 errorHandler 来提供 KafkaListenerErrorHandler 实现的 Bean 名称。这个功能接口有一个方法，如下所示：

@FunctionalInterface
public interface KafkaListenerErrorHandler {
    Object handleError(Message<?> message, ListenerExecutionFailedException exception) throws Exception;

你可以访问消息转换器产生的 Spring-messaging Message<?> 对象，以及侦听器抛出的异常，该异常包装在 ListenerExecutionFailedException 中。错误处理程序可以抛出原始异常或新的异常，这些异常将被抛出到容器中。错误处理程序返回的任何内容都将被忽略。

从版本 2.7 开始，你可以在 MessagingMessageConverter 和 BatchMessagingMessageConverter 上设置 rawRecordHeader 属性，这会导致将 RAW ConsumerRecord 添加到 KafkaHeaders.RAW_DATA 标头中转换的 Message<?> 中。这是有用的，例如，如果你希望在侦听器错误处理程序中使用 DeadLetterPublishingRecoverer 。它可能用于请求/回复场景，在此场景中，你希望在重试一定次数后，在捕获死信主题中的失败记录后，将失败结果发送给发件人。

@Bean
KafkaListenerErrorHandler eh(DeadLetterPublishingRecoverer recoverer) {
    return (msg, ex) -> {
        if (msg.getHeaders().get(KafkaHeaders.DELIVERY_ATTEMPT, Integer.class) > 9) {
            recoverer.accept(msg.getHeaders().get(KafkaHeaders.RAW_DATA, ConsumerRecord.class), ex);
            return "FAILED";
        throw ex;

它有一个子接口（ ConsumerAwareListenerErrorHandler ），可以通过以下方法访问消费者对象：

Object handleError(Message<?> message, ListenerExecutionFailedException exception, Consumer<?, ?> consumer);

如果你的错误处理程序实现了这个接口，那么你可以（例如）相应地调整偏移量。例如，要重置偏移量以重播失败的消息，你可以执行以下操作：

@Bean
public ConsumerAwareListenerErrorHandler listen3ErrorHandler() {
    return (m, e, c) -> {
        this.listen3Exception = e;
        MessageHeaders headers = m.getHeaders();
        c.seek(new org.apache.kafka.common.TopicPartition(
                headers.get(KafkaHeaders.RECEIVED_TOPIC, String.class),
                headers.get(KafkaHeaders.RECEIVED_PARTITION_ID, Integer.class)),
                headers.get(KafkaHeaders.OFFSET, Long.class));
        return null;

类似地，你可以为批处理侦听器执行如下操作：

@Bean
public ConsumerAwareListenerErrorHandler listen10ErrorHandler() {
    return (m, e, c) -> {
        this.listen10Exception = e;
        MessageHeaders headers = m.getHeaders();
        List<String> topics = headers.get(KafkaHeaders.RECEIVED_TOPIC, List.class);
        List<Integer> partitions = headers.get(KafkaHeaders.RECEIVED_PARTITION_ID, List.class);
        List<Long> offsets = headers.get(KafkaHeaders.OFFSET, List.class);
        Map<TopicPartition, Long> offsetsToReset = new HashMap<>();
        for (int i = 0; i < topics.size(); i++) {
            int index = i;
            offsetsToReset.compute(new TopicPartition(topics.get(i), partitions.get(i)),
                    (k, v) -> v == null ? offsets.get(index) : Math.min(v, offsets.get(index)));
        offsetsToReset.forEach((k, v) -> c.seek(k, v));
        return null;

这会将批处理中的每个主题/分区重置为批处理中的最低偏移量。

# 容器错误处理程序

从版本 2.8 开始，遗留的 ErrorHandler 和 BatchErrorHandler 接口已被一个新的 CommonErrorHandler 所取代。这些错误处理程序可以同时处理记录和批处理侦听器的错误，从而允许单个侦听器容器工厂为这两种类型的侦听器创建容器。 CommonErrorHandler 替换大多数遗留框架错误处理程序的实现被提供，并且不推荐遗留错误处理程序。遗留接口仍然受到侦听器容器和侦听器容器工厂的支持；它们将在未来的版本中被弃用。

在使用事务时，默认情况下不会配置错误处理程序，因此异常将回滚事务。事务容器的错误处理由[ AfterRollbackProcessor ]（#after-rollback）处理。如果你在使用事务时提供了自定义错误处理程序，那么如果你希望回滚事务，它必须抛出异常。

这个接口有一个默认的方法 isAckAfterHandle() ，容器调用它来确定如果错误处理程序返回而没有抛出异常，是否应该提交偏移量；默认情况下，它返回 true。

通常，框架提供的错误处理程序将在错误未得到“处理”时（例如，在执行查找操作后）抛出异常。默认情况下，此类异常由容器在 ERROR 级别记录。所有框架错误处理程序都扩展了 KafkaExceptionLogLevelAware ，它允许你控制记录这些异常的级别。

/**
 * Set the level at which the exception thrown by this handler is logged.
 * @param logLevel the level (default ERROR).
public void setLogLevel(KafkaException.Level logLevel) {

你可以为容器工厂中的所有侦听器指定一个全局错误处理程序。下面的示例展示了如何做到这一点：

@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
        kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
    factory.setCommonErrorHandler(myErrorHandler);
    return factory;

默认情况下，如果带注释的侦听器方法抛出异常，则将其抛出到容器中，并根据容器配置来处理消息。

容器在调用错误处理程序之前提交任何挂起的偏移量提交。

如果使用 Spring 引导，只需将错误处理程序添加为 @Bean ，然后引导将其添加到自动配置的工厂。

# DefaulTerrorHandler

这个新的错误处理程序替换了 SeekToCurrentErrorHandler 和 RecoveringBatchErrorHandler ，它们现在已经是几个版本的默认错误处理程序。一个不同之处是批处理侦听器的回退行为（当抛出 BatchListenerFailedException 以外的异常时）与 重试完整批 是等价的。

错误处理程序可以恢复（跳过）持续失败的记录。默认情况下，在十次失败之后，将记录失败的记录（在 ERROR 级别）。你可以使用一个自定义的 recoverer（ BiConsumer ）和一个 BackOff 来配置处理程序，该 receverer 和 BackOff 控制每次交付的尝试和延迟。使用 FixedBackOff 和 FixedBackOff.UNLIMITED_ATTEMPTS 可以（有效地）导致无限次重试。下面的示例在三次尝试后配置恢复：

DefaultErrorHandler errorHandler =
    new DefaultErrorHandler((record, exception) -> {
        // recover after 3 failures, with no back off - e.g. send to a dead-letter topic
    }, new FixedBackOff(0L, 2L));

要用此处理程序的定制实例配置侦听器容器，请将其添加到容器工厂。

例如，对于 @KafkaListener 容器工厂，你可以添加 DefaultErrorHandler ，如下所示：

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory();
    factory.setConsumerFactory(consumerFactory());
    factory.getContainerProperties().setAckOnError(false);
    factory.getContainerProperties().setAckMode(AckMode.RECORD);
    factory.setCommonErrorHandler(new DefaultErrorHandler(new FixedBackOff(1000L, 2L)));
    return factory;

对于记录侦听器，这将重试一次交付多达 2 次（3 次交付尝试），并后退 1 秒，而不是默认配置（ FixedBackOff(0L, 9) ）。在重试结束后，只需记录失败的次数。

例如；如果 poll 返回六条记录（每个分区 0、1、2 有两条记录），并且侦听器在第四条记录上抛出异常，则容器通过提交它们的偏移量来确认前三条消息。 DefaultErrorHandler 寻求分区 1 的偏移量 1 和分区 2 的偏移量 0.下一个 poll() 返回这三条未处理的记录。

如果 AckMode 是 BATCH ，则容器在调用错误处理程序之前提交前两个分区的偏移量。

对于批处理侦听器，侦听器必须抛出 BatchListenerFailedException ，指示批处理中的哪些记录失败。

事件的顺序是：

• 在索引之前提交记录的偏移量。

• 如果没有用尽重试，则执行查找，以便将所有剩余的记录（包括失败的记录）重新交付。

• 如果重复尝试已用尽，请尝试恢复失败的记录（仅缺省日志）并执行查找，以便重新交付剩余的记录（不包括失败的记录）。已提交已恢复记录的偏移量。

• 如果重试已用尽，而恢复失败，则进行查找，就好像重试尚未用尽一样。

在重试结束后，默认的 recoverer 会记录失败的记录。你可以使用自定义的 recoverer，或者由框架提供的一个，例如[ DeadLetterPublishingRecoverer ]（#dead-letters）。

当使用 POJO 批处理侦听器（例如 List<Thing> ）时，如果没有完整的消费者记录可添加到异常中，则只需添加失败记录的索引：

@KafkaListener(id = "recovering", topics = "someTopic")
public void listen(List<Thing> things) {
    for (int i = 0; i < records.size(); i++) {
        try {
            process(things.get(i));
        catch (Exception e) {
            throw new BatchListenerFailedException("Failed to process", i);

当容器配置为 AckMode.MANUAL_IMMEDIATE 时，可以将错误处理程序配置为提交恢复记录的偏移量；将 commitRecovered 属性设置为 true 。

另见 发布死信记录 。

当使用事务时，类似的功能由 DefaultAfterRollbackProcessor 提供。见 后回滚处理器 。

DefaultErrorHandler 认为某些异常是致命的，对于此类异常跳过重试；在第一次失败时调用 recuverer。默认情况下，被认为是致命的例外是：

• DeserializationException

• MessageConversionException

• ConversionException

• MethodArgumentResolutionException

• NoSuchMethodException

• ClassCastException

因为这些异常不太可能在重试交付时得到解决。

你可以将更多的异常类型添加到不可重排的类别中，或者完全替换分类异常的映射。有关更多信息，请参见 DefaultErrorHandler.addNotRetryableException() 和 DefaultErrorHandler.setClassifications() 的 Javadocs，以及 spring-retry``BinaryExceptionClassifier 的 Javadocs。

下面是一个将 IllegalArgumentException 添加到不可重排异常的示例：

@Bean
public DefaultErrorHandler errorHandler(ConsumerRecordRecoverer recoverer) {
    DefaultErrorHandler handler = new DefaultErrorHandler(recoverer);
    handler.addNotRetryableExceptions(IllegalArgumentException.class);
    return handler;

可以将错误处理程序配置为一个或多个 RetryListener s，接收重试和恢复进度的通知。

@FunctionalInterface
public interface RetryListener {
    void failedDelivery(ConsumerRecord<?, ?> record, Exception ex, int deliveryAttempt);
    default void recovered(ConsumerRecord<?, ?> record, Exception ex) {
    default void recoveryFailed(ConsumerRecord<?, ?> record, Exception original, Exception failure) {

有关更多信息，请参见 Javadocs。

你可以向错误处理程序提供 BiFunction<ConsumerRecord<?, ?>, Exception, BackOff> ，以基于失败的记录和/或异常来确定要使用的 BackOff :

handler.setBackOffFunction((record, ex) -> { ... });

如果函数返回 null ，将使用处理程序的默认 BackOff 。

将 resetStateOnExceptionChange 设置为 true ，如果异常类型在两次失败之间发生变化，则将重新启动重试序列（包括选择新的 BackOff ，如果这样配置的话）。默认情况下，不考虑异常类型。

另见 传递尝试标头 。

# 4.1.21.使用批处理错误处理程序的转换错误

从版本 2.8 开始，批处理侦听器现在可以正确处理转换错误，当使用 MessageConverter 和 ByteArrayDeserializer 、 BytesDeserializer 或 StringDeserializer 以及 DefaultErrorHandler 时。当发生转换错误时，将有效负载设置为 null，并将反序列化异常添加到记录头中，类似于 ErrorHandlingDeserializer 。侦听器中有一个 ConversionException s 的列表可用，因此侦听器可以抛出一个 BatchListenerFailedException ，指示发生转换异常的第一个索引。

示例：

@KafkaListener(id = "test", topics = "topic")
void listen(List<Thing> in, @Header(KafkaHeaders.CONVERSION_FAILURES) List<ConversionException> exceptions) {
    for (int i = 0; i < in.size(); i++) {
        Foo foo = in.get(i);
        if (foo == null && exceptions.get(i) != null) {
            throw new BatchListenerFailedException("Conversion error", exceptions.get(i), i);
        process(foo);

# 重试完整批

这就是批侦听器 DefaultErrorHandler 的回退行为，其中侦听器抛出一个 BatchListenerFailedException 以外的异常。

不能保证当一个批被重新交付时，该批具有相同数量的记录和/或重新交付的记录的顺序相同。因此，不可能轻松地保持批处理的重试状态。 FallbackBatchErrorHandler 采取如下方法。如果批处理侦听器抛出一个不是 BatchListenerFailedException 的异常，则从内存中的批记录执行重试。为了避免在扩展的重试过程中发生再平衡，错误处理程序会暂停使用者，在每次重试之前对其进行轮询，并再次调用侦听器。如果/当重试用完时，将为批处理中的每个记录调用 ConsumerRecordRecoverer 。如果 Recoverer 抛出一个异常，或者线程在睡眠期间被中断，则该批记录将在下一次投票时重新交付。在退出之前，无论结果如何，消费者都会被恢复。

在等待 BackOff 间隔期间，错误处理程序将进行短暂的休眠循环，直到达到所需的延迟，同时检查容器是否已停止，从而允许在 stop() 之后不久退出休眠，而不是导致延迟。

# 容器停止错误处理程序

如果侦听器抛出异常， CommonContainerStoppingErrorHandler 将停止容器。对于记录侦听器，当 AckMode 为 RECORD 时，将提交已处理记录的偏移。对于记录侦听器，当 AckMode 是任意手动值时，将提交已确认记录的偏移量。对于记录侦听器，当 AckMode 是 BATCH 时，或者对于批处理侦听器，当容器重新启动时，整个批处理将被重新播放。

在容器停止之后，抛出一个包装 ListenerExecutionFailedException 的异常。这将导致事务回滚（如果启用了事务）。

# 委派错误处理程序

根据异常类型的不同， CommonDelegatingErrorHandler 可以委托给不同的错误处理程序。例如，你可能希望对大多数异常调用 DefaultErrorHandler ，或者对其他异常调用 CommonContainerStoppingErrorHandler 。

# 日志错误处理程序

CommonLoggingErrorHandler 只记录异常；使用记录侦听器，上一次投票的剩余记录将传递给侦听器。对于批处理侦听器，将记录批处理中的所有记录。

# 对记录和批处理侦听器使用不同的常见错误处理程序

如果你希望对记录和批处理侦听器使用不同的错误处理策略，则提供 CommonMixedErrorHandler ，允许为每个侦听器类型配置特定的错误处理程序。

# 常见错误处理程序 summery

• DefaultErrorHandler

• CommonContainerStoppingErrorHandler

• CommonDelegatingErrorHandler

• CommonLoggingErrorHandler

• CommonMixedErrorHandler

# 遗留错误处理程序及其替换程序

# 后回滚处理器

在使用事务时，如果侦听器抛出一个异常（如果存在错误处理程序，则抛出一个异常），事务将被回滚。默认情况下，任何未处理的记录（包括失败的记录）都会在下一次投票时重新获取。这是通过在 DefaultAfterRollbackProcessor 中执行 seek 操作来实现的。使用批处理侦听器，整个批记录将被重新处理（容器不知道批处理中的哪一条记录失败了）。要修改此行为，可以使用自定义 AfterRollbackProcessor 配置侦听器容器。例如，对于基于记录的侦听器，你可能希望跟踪失败的记录，并在尝试了一定次数后放弃，也许可以将其发布到一个死信不疑的主题中。

从版本 2.2 开始， DefaultAfterRollbackProcessor 现在可以恢复（跳过）一条持续失败的记录。默认情况下，在十次失败之后，将记录失败的记录（在 ERROR 级别）。你可以使用自定义的 recoverer（ BiConsumer ）和最大故障来配置处理器。将 maxFailures 属性设置为负数会导致无限次重试。下面的示例在三次尝试后配置恢复：

AfterRollbackProcessor<String, String> processor =
    new DefaultAfterRollbackProcessor((record, exception) -> {
        // recover after 3 failures, with no back off - e.g. send to a dead-letter topic
    }, new FixedBackOff(0L, 2L));

当不使用事务时，可以通过配置 DefaultErrorHandler 来实现类似的功能。见 容器错误处理程序 。

另见 发布死信记录 。

从版本 2.2.5 开始，可以在新事务中调用 DefaultAfterRollbackProcessor （在失败的事务回滚后启动）。然后，如果你使用 DeadLetterPublishingRecoverer 来发布失败的记录，处理器将把恢复的记录在原始主题/分区中的偏移量发送给事务。要启用此功能，请在 DefaultAfterRollbackProcessor 上设置 commitRecovered 和 kafkaTemplate 属性。

从版本 2.6 开始，你现在可以为处理器提供一个 BiFunction<ConsumerRecord<?, ?>, Exception, BackOff> ，以基于失败的记录和/或异常来确定要使用的 BackOff :

handler.setBackOffFunction((record, ex) -> { ... });

如果函数返回 null ，将使用处理器的默认 BackOff 。

从版本 2.6.3 开始，将 resetStateOnExceptionChange 设置为 true ，如果异常类型在两次失败之间发生变化，则将重新启动重试序列（包括选择一个新的 BackOff ，如果这样配置的话）。默认情况下，不考虑异常类型。

从版本 2.3.1 开始，类似于 DefaultErrorHandler ， DefaultAfterRollbackProcessor 认为某些异常是致命的，并且对于此类异常跳过重试；在第一次失败时调用 recoverer。默认情况下，被认为是致命的例外是：

• DeserializationException

• MessageConversionException

• ConversionException

• MethodArgumentResolutionException

• NoSuchMethodException

• ClassCastException

因为这些异常不太可能在重试交付时得到解决。

你可以将更多的异常类型添加到不可重排的类别中，或者完全替换分类异常的映射。有关更多信息，请参见 DefaultAfterRollbackProcessor.setClassifications() 的 Javadocs，以及 spring-retry``BinaryExceptionClassifier 的 Javadocs。

下面是一个将 IllegalArgumentException 添加到不可重排异常的示例：

@Bean
public DefaultAfterRollbackProcessor errorHandler(BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer) {
    DefaultAfterRollbackProcessor processor = new DefaultAfterRollbackProcessor(recoverer);
    processor.addNotRetryableException(IllegalArgumentException.class);
    return processor;

另见 传递尝试标头 。

从版本 2.7 开始，在等待 BackOff 间隔期间，错误处理程序将进行短暂的休眠循环，直到达到所需的延迟，同时检查容器是否已停止，允许睡眠在 stop() 后很快退出，而不是导致延迟。

从版本 2.7 开始，处理器可以配置一个或多个 RetryListener s，接收重试和恢复进度的通知。

@FunctionalInterface
public interface RetryListener {
    void failedDelivery(ConsumerRecord<?, ?> record, Exception ex, int deliveryAttempt);
    default void recovered(ConsumerRecord<?, ?> record, Exception ex) {
    default void recoveryFailed(ConsumerRecord<?, ?> record, Exception original, Exception failure) {

有关更多信息，请参见 Javadocs。

# 投递尝试头

以下内容仅适用于记录侦听器，而不是批处理侦听器。

从版本 2.5 开始，当使用实现 DeliveryAttemptAware 或 AfterRollbackProcessor 的 AfterRollbackProcessor 时，可以在记录中添加 KafkaHeaders.DELIVERY_ATTEMPT 头（ kafka_deliveryAttempt ）。这个标头的值是一个从 1 开始的递增整数。当接收到 RAW ConsumerRecord<?, ?> 时，该整数在 byte[4] 中。

int delivery = ByteBuffer.wrap(record.headers()
    .lastHeader(KafkaHeaders.DELIVERY_ATTEMPT).value())
    .getInt()

当将 @KafkaListener 与 DefaultKafkaHeaderMapper 或 SimpleKafkaHeaderMapper 一起使用时，可以通过将 @Header(KafkaHeaders.DELIVERY_ATTEMPT) int delivery 作为参数添加到侦听器方法中来获得。

要启用这个头的填充，将容器属性 deliveryAttemptHeader 设置为 true 。默认情况下禁用它，以避免查找每个记录的状态并添加标题的（小）开销。

DefaultErrorHandler 和 DefaultAfterRollbackProcessor 支持此功能。

# 发布死信记录

当某项记录的失败次数达到最大值时，可以使用记录恢复程序配置 DefaultErrorHandler 和 DefaultAfterRollbackProcessor 。该框架提供 DeadLetterPublishingRecoverer ，用于将失败的消息发布到另一个主题。recoverer 需要一个 KafkaTemplate<Object, Object> ，用于发送记录。你还可以选择用 BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> 配置它，调用它是为了解析目标主题和分区。

如果返回的 TopicPartition 有一个负分区，则该分区未在 ProducerRecord 中设置，因此该分区由 Kafka 选择。从版本 2.2.4 开始，任何 ListenerExecutionFailedException （例如，当在 @KafkaListener 方法中检测到异常时抛出）都将使用 groupId 属性进行增强。这允许目标解析器使用这个，除了在 ConsumerRecord 中选择死信主题的信息之外。

下面的示例展示了如何连接自定义目标解析器：

DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template,
        (r, e) -> {
            if (e instanceof FooException) {
                return new TopicPartition(r.topic() + ".Foo.failures", r.partition());
            else {
                return new TopicPartition(r.topic() + ".other.failures", r.partition());
ErrorHandler errorHandler = new DefaultErrorHandler(recoverer, new FixedBackOff(0L, 2L));

发送到死信主题的记录通过以下标题进行了增强：

• KafkaHeaders.DLT_EXCEPTION_FQCN :异常类名称（一般是 ListenerExecutionFailedException ，但也可以是其他的）。

• KafkaHeaders.DLT_EXCEPTION_CAUSE_FQCN :异常导致类名，如果存在的话（自版本 2.8 起）。

• KafkaHeaders.DLT_EXCEPTION_STACKTRACE :异常堆栈跟踪。

• KafkaHeaders.DLT_EXCEPTION_MESSAGE :异常消息。

• KafkaHeaders.DLT_KEY_EXCEPTION_FQCN :异常类名（仅限键反序列化错误）。

• KafkaHeaders.DLT_KEY_EXCEPTION_STACKTRACE :异常堆栈跟踪（仅限键反序列化错误）。

• KafkaHeaders.DLT_KEY_EXCEPTION_MESSAGE :异常消息（仅限键反序列化错误）。

• KafkaHeaders.DLT_ORIGINAL_TOPIC :原始主题。

• KafkaHeaders.DLT_ORIGINAL_PARTITION :原始分区。

• KafkaHeaders.DLT_ORIGINAL_OFFSET :原始偏移量。

• KafkaHeaders.DLT_ORIGINAL_TIMESTAMP :原始时间戳。

• KafkaHeaders.DLT_ORIGINAL_TIMESTAMP_TYPE :原始时间戳类型。

• KafkaHeaders.DLT_ORIGINAL_CONSUMER_GROUP :未能处理记录的原始消费者组（自版本 2.8 起）。

关键异常仅由 DeserializationException s 引起，因此不存在 DLT_KEY_EXCEPTION_CAUSE_FQCN 。

有两种机制可以添加更多的头。

1. 子类 recoverer 和 override createProducerRecord() -调用 super.createProducerRecord() 并添加更多标题。

2. 提供一个 BiFunction 来接收消费者记录和异常，返回一个 Headers 对象；头从那里将被复制到最终的生产者记录。使用 setHeadersFunction() 设置 BiFunction 。

第二种方法实现起来更简单，但第一种方法有更多可用信息，包括已经组装好的标准标头。

从版本 2.3 开始，当与 ErrorHandlingDeserializer 一起使用时，发布者将把死信生成器记录中的记录 value() 恢复到无法反序列化的原始值。以前， value() 是空的，用户代码必须从消息头中解码 DeserializationException 。此外，你还可以向发布者提供多个 KafkaTemplate s；这可能是必需的，例如，如果你希望从 DeserializationException 中发布 byte[] ，以及使用不同的序列化器从已成功反序列化的记录中发布的值。下面是一个使用 KafkaTemplate s 和 byte[] 序列化器配置发布服务器的示例：

@Bean
public DeadLetterPublishingRecoverer publisher(KafkaTemplate<?, ?> stringTemplate,
        KafkaTemplate<?, ?> bytesTemplate) {
    Map<Class<?>, KafkaTemplate<?, ?>> templates = new LinkedHashMap<>();
    templates.put(String.class, stringTemplate);
    templates.put(byte[].class, bytesTemplate);
    return new DeadLetterPublishingRecoverer(templates);

发布者使用映射键来定位适合即将发布的 value() 的模板。建议使用 LinkedHashMap ，以便按顺序检查密钥。

当发布 null 值时，当有多个模板时，recoverer 将为 Void 类寻找一个模板；如果不存在，将使用 values().iterator() 中的第一个模板。

从 2.7 开始，你可以使用 setFailIfSendResultIsError 方法，以便在消息发布失败时引发异常。你还可以使用 setWaitForSendResultTimeout 设置用于验证发送方成功的超时。

从版本 2.6.3 开始，将 resetStateOnExceptionChange 设置为 true ，如果异常类型在两次失败之间发生变化，则将重新启动重试序列（包括选择一个新的 BackOff ，如果这样配置的话）。默认情况下，不考虑异常类型。

从版本 2.3 开始，Recoverer 还可以与 Kafka Streams 一起使用-有关更多信息，请参见 从反序列化异常恢复 。

ErrorHandlingDeserializer 在头 ErrorHandlingDeserializer.VALUE_DESERIALIZER_EXCEPTION_HEADER 和 ErrorHandlingDeserializer.KEY_DESERIALIZER_EXCEPTION_HEADER （使用 Java 序列化）中添加了反序列化异常。默认情况下，这些标题不会保留在发布到死信主题的消息中。从版本 2.7 开始，如果键和值都反序列化失败，那么这两个键的原始值都会在发送到 DLT 的记录中填充。

如果接收到的记录是相互依赖的，但可能会到达顺序错误，那么将失败的记录重新发布到原始主题的尾部（多次）可能会很有用，而不是直接将其发送到死信主题。例如，见 这个堆栈溢出问题 (opens new window) 。

下面的错误处理程序配置将完全做到这一点：

@Bean
public ErrorHandler eh(KafkaOperations<String, String> template) {
    return new DefaultErrorHandler(new DeadLetterPublishingRecoverer(template,
            (rec, ex) -> {
                org.apache.kafka.common.header.Header retries = rec.headers().lastHeader("retries");
                if (retries == null) {
                    retries = new RecordHeader("retries", new byte[] { 1 });
                    rec.headers().add(retries);
                else {
                    retries.value()[0]++;
                return retries.value()[0] > 5
                        ? new TopicPartition("topic.DLT", rec.partition())
                        : new TopicPartition("topic", rec.partition());
            }), new FixedBackOff(0L, 0L));

从版本 2.7 开始，recoverer 将检查目标解析程序选择的分区是否确实存在。如果不存在分区，则将 ProducerRecord 中的分区设置为 null ，从而允许 KafkaProducer 选择该分区。可以通过将 verifyPartition 属性设置为 false 来禁用此检查。

# 管理死信记录头

参考上面的 发布死信记录 ， DeadLetterPublishingRecoverer 有两个属性，当这些头已经存在时（例如，当重新处理失败的死信记录时，包括使用 非阻塞重试 时），这些属性用于管理头。

• appendOriginalHeaders （默认 true ）

• stripPreviousExceptionHeaders （默认 true 自 2.8 版本）

Apache Kafka 支持同名的多个头；要获得“latest”值，可以使用 headers.lastHeader(headerName) ；要在多个头上获得迭代器，可以使用 headers.headers(headerName).iterator() 。

当重复重新发布失败的记录时，这些标头可能会增加（并最终由于 RecordTooLargeException 而导致发布失败）；对于异常标头，尤其是对于堆栈跟踪标头，尤其如此。

产生这两个属性的原因是，虽然你可能只希望保留最后一个异常信息，但你可能希望保留记录在每次失败时通过的主题的历史记录。

appendOriginalHeaders 应用于所有名为 **ORIGINAL** 的标头，而 stripPreviousExceptionHeaders 应用于所有名为 **EXCEPTION** 的标头。

另见 故障报头管理 与 非阻塞重试 。

# ExponentialBackOffWithMaxRetries 实现

Spring 框架提供了许多 BackOff 实现方式。默认情况下， ExponentialBackOff 将无限期地重试；如果要在多次重试后放弃，则需要计算 maxElapsedTime 。由于版本 2.7.3， Spring for Apache Kafka 提供了 ExponentialBackOffWithMaxRetries ，这是一个子类，它接收 maxRetries 属性并自动计算 maxElapsedTime ，这更方便一些。

@Bean
DefaultErrorHandler handler() {
    ExponentialBackOffWithMaxRetries bo = new ExponentialBackOffWithMaxRetries(6);
    bo.setInitialInterval(1_000L);
    bo.setMultiplier(2.0);
    bo.setMaxInterval(10_000L);
    return new DefaultErrorHandler(myRecoverer, bo);

这将在 1, 2, 4, 8, 10, 10 秒后重试，然后再调用 recoverer。

# 4.1.22.Jaas 和 Kerberos

从版本 2.0 开始，添加了一个 KafkaJaasLoginModuleInitializer 类来帮助 Kerberos 配置。你可以使用所需的配置将这个 Bean 添加到你的应用程序上下文中。下面的示例配置了这样的 Bean:

@Bean
public KafkaJaasLoginModuleInitializer jaasConfig() throws IOException {
    KafkaJaasLoginModuleInitializer jaasConfig = new KafkaJaasLoginModuleInitializer();
    jaasConfig.setControlFlag("REQUIRED");
    Map<String, String> options = new HashMap<>();
    options.put("useKeyTab", "true");
    options.put("storeKey", "true");
    options.put("keyTab", "/etc/security/keytabs/kafka_client.keytab");
    options.put("principal", "[email protected]");
    jaasConfig.setOptions(options);
    return jaasConfig;

# 4.2. Apache Kafka Streams 支持

从版本 1.1.4 开始， Spring for Apache Kafka 为 Kafka溪流 (opens new window) 提供了一流的支持。要在 Spring 应用程序中使用它， kafka-streams jar 必须存在于 Classpath 上。它是 Spring for Apache Kafka 项目的可选依赖项，并且不是通过传递方式下载的。

# 4.2.1.基础知识

参考文献 Apache Kafka Streams 文档建议使用以下 API 的方式：

// Use the builders to define the actual processing topology, e.g. to specify
// from which input topics to read, which stream operations (filter, map, etc.)
// should be called, and so on.
StreamsBuilder builder = ...;  // when using the Kafka Streams DSL
// Use the configuration to tell your application where the Kafka cluster is,
// which serializers/deserializers to use by default, to specify security settings,
// and so on.
StreamsConfig config = ...;
KafkaStreams streams = new KafkaStreams(builder, config);
// Start the Kafka Streams instance
streams.start();
// Stop the Kafka Streams instance
streams.close();

因此，我们有两个主要组成部分：

• StreamsBuilder :使用 API 构建 KStream （或 KTable ）实例。

• KafkaStreams :管理这些实例的生命周期。

# 4.2.2. Spring 管理

为了简化从 Spring 应用程序上下文视角使用 Kafka 流并通过容器使用生命周期管理， Spring for Apache Kafka 引入了 StreamsBuilderFactoryBean 。这是一个 AbstractFactoryBean 实现，用于将 StreamsBuilder 单例实例公开为 Bean。下面的示例创建了这样的 Bean:

@Bean
public FactoryBean<StreamsBuilder> myKStreamBuilder(KafkaStreamsConfiguration streamsConfig) {
    return new StreamsBuilderFactoryBean(streamsConfig);

StreamsBuilderFactoryBean 还实现了 SmartLifecycle 来管理内部 KafkaStreams 实例的生命周期。与 Kafka Streams API 类似，在启动 KafkaStreams 之前，必须定义 KStream 实例。这也适用于 Kafka Streams 的 Spring API。因此，当你在 StreamsBuilderFactoryBean 上使用默认的 autoStartup = true 时，你必须在刷新应用程序上下文之前在 KStream 上声明 KStream 实例。例如， KStream 可以是一个常规的 Bean 定义，而 Kafka Streams API 的使用没有任何影响。下面的示例展示了如何做到这一点：

@Bean
public KStream<?, ?> kStream(StreamsBuilder kStreamBuilder) {
    KStream<Integer, String> stream = kStreamBuilder.stream(STREAMING_TOPIC1);
    // Fluent KStream API
    return stream;

如果希望手动控制生命周期（例如，通过某些条件停止和启动），则可以通过使用工厂 Bean（ StreamsBuilderFactoryBean ）直接引用 prefix (opens new window) 。由于 StreamsBuilderFactoryBean 使用其内部 KafkaStreams 实例，因此可以安全地停止并重新启动它。在每个 start() 上创建一个新的 KafkaStreams 。如果你希望单独控制 KStream 实例的生命周期，那么你也可以考虑使用不同的 StreamsBuilderFactoryBean 实例。

你还可以在 StreamsBuilderFactoryBean 上指定 KafkaStreams.StateListener 、 Thread.UncaughtExceptionHandler 和 StateRestoreListener 选项，这些选项被委托给内部 KafkaStreams 实例。此外，除了在 StreamsBuilderFactoryBean 上以 版本 2.1.5 间接设置这些选项外，还可以使用 KafkaStreams 回调接口来配置内部 KafkaStreams 实例。请注意， KafkaStreamsCustomizer 覆盖了 StreamsBuilderFactoryBean 提供的选项。如果需要直接执行一些 KafkaStreams 操作，则可以使用 StreamsBuilderFactoryBean.getKafkaStreams() 访问内部 KafkaStreams 实例。可以按类型自动连接 StreamsBuilderFactoryBean Bean，但应确保在 Bean 定义中使用完整的类型，如下例所示：

@Bean
public StreamsBuilderFactoryBean myKStreamBuilder(KafkaStreamsConfiguration streamsConfig) {
    return new StreamsBuilderFactoryBean(streamsConfig);
@Autowired
private StreamsBuilderFactoryBean myKStreamBuilderFactoryBean;

或者，如果使用接口 Bean 定义，则可以按名称添加 @Qualifier 用于注入。下面的示例展示了如何做到这一点：

@Bean
public FactoryBean<StreamsBuilder> myKStreamBuilder(KafkaStreamsConfiguration streamsConfig) {
    return new StreamsBuilderFactoryBean(streamsConfig);
@Autowired
@Qualifier("&myKStreamBuilder")
private StreamsBuilderFactoryBean myKStreamBuilderFactoryBean;

从版本 2.4.1 开始，工厂 Bean 有一个新的属性 infrastructureCustomizer ，类型为 KafkaStreamsInfrastructureCustomizer ；这允许在创建流之前自定义 StreamsBuilder （例如添加状态存储）和/或 Topology 。

public interface KafkaStreamsInfrastructureCustomizer {
	void configureBuilder(StreamsBuilder builder);
	void configureTopology(Topology topology);

提供了默认的无操作实现，以避免在不需要两个方法的情况下不得不实现这两个方法。

提供了一个 CompositeKafkaStreamsInfrastructureCustomizer ，用于在需要应用多个自定义程序时。

# 4.2.3.Kafkastreams 测微仪支持

在版本 2.5.3 中引入的，可以配置 KafkaStreamsMicrometerListener 来为工厂 Bean 管理的 KafkaStreams 对象自动注册千分表：

streamsBuilderFactoryBean.addListener(new KafkaStreamsMicrometerListener(meterRegistry,
        Collections.singletonList(new ImmutableTag("customTag", "customTagValue"))));

# 4.2.4.流 JSON 序列化和反序列化

对于在以 JSON 格式读取或写入主题或状态存储时序列化和反序列化数据， Spring for Apache Kafka 提供了一个 JsonSerde 实现，该实现使用 JSON，将其委托给 JsonSerializer 和 JsonDeserializer 中描述的 序列化、反序列化和消息转换 。 JsonSerde 实现通过其构造函数（目标类型或 ObjectMapper ）提供相同的配置选项。在下面的示例中，我们使用 JsonSerde 序列化和反序列化 Kafka 流的 Cat 有效负载（只要需要实例， JsonSerde 就可以以类似的方式使用）：

stream.through(Serdes.Integer(), new JsonSerde<>(Cat.class), "cats");

从版本 2.3 开始，当以编程方式构建在生产者/消费者工厂中使用的序列化器/反序列化器时，你可以使用 Fluent API，这简化了配置。

stream.through(new JsonSerde<>(MyKeyType.class)
        .forKeys()
        .noTypeInfo(),
    new JsonSerde<>(MyValueType.class)
        .noTypeInfo(),
    "myTypes");

# 4.2.5.使用 KafkaStreamBrancher

KafkaStreamBrancher 类引入了一种在 KStream 之上构建条件分支的更方便的方法。

考虑以下不使用 KafkaStreamBrancher 的示例：

KStream<String, String>[] branches = builder.stream("source").branch(
      (key, value) -> value.contains("A"),
      (key, value) -> value.contains("B"),
      (key, value) -> true
branches[0].to("A");
branches[1].to("B");
branches[2].to("C");

下面的示例使用 KafkaStreamBrancher :

new KafkaStreamBrancher<String, String>()
   .branch((key, value) -> value.contains("A"), ks -> ks.to("A"))
   .branch((key, value) -> value.contains("B"), ks -> ks.to("B"))
   //default branch should not necessarily be defined in the end of the chain!
   .defaultBranch(ks -> ks.to("C"))
   .onTopOf(builder.stream("source"));
   //onTopOf method returns the provided stream so we can continue with method chaining

# 4.2.6.配置

要配置 Kafka Streams 环境， StreamsBuilderFactoryBean 需要一个 KafkaStreamsConfiguration 实例。有关所有可能的选项，请参见 Apache kafka 文件 (opens new window) 。

为了避免在大多数情况下使用样板代码，特别是在开发微服务时， Spring for Apache Kafka 提供了 @EnableKafkaStreams 注释，你应该将其放置在 @Configuration 类上。只需要声明一个名为 KafkaStreamsConfiguration Bean 的 defaultKafkaStreamsConfig 。在应用程序上下文中自动声明一个名为 StreamsBuilderFactoryBean Bean 的 defaultKafkaStreamsBuilder 。你也可以声明和使用任何额外的 StreamsBuilderFactoryBean bean。通过提供实现 StreamsBuilderFactoryBeanConfigurer 的 Bean，你可以对该 Bean 执行额外的自定义。如果有多个这样的 bean，则将根据其 Ordered.order 属性应用它们。

默认情况下，当工厂 Bean 停止时，将调用 KafkaStreams.cleanUp() 方法。从版本 2.1.2 开始，工厂 Bean 有额外的构造函数，接受一个 CleanupConfig 对象，该对象具有属性，可以让你控制在 cleanUp() 或 stop() 期间是否调用 cleanUp() 方法。从版本 2.7 开始，默认情况是永远不清理本地状态。

# 4.2.7.页眉 Enricher

版本 2.3 增加了 HeaderEnricher 的 Transformer 实现。这可用于在流处理中添加头；头的值是 SPEL 表达式；表达式求值的根对象具有 3 个属性：

• context - ProcessorContext ，允许访问当前记录的元数据

• key -当前记录的键

• value -当前记录的值

表达式必须返回 byte[] 或 String （使用 UTF-8 将其转换为 byte[] ）。

要在流中使用 Enrich:

.transform(() -> enricher)

转换器不改变 key 或 value ；它只是添加了标题。

.transform(() -> new HeaderEnricher<..., ...>(expressionMap))

下面是一个简单的示例，添加了一个文字头和一个变量：

Map<String, Expression> headers = new HashMap<>();
headers.put("header1", new LiteralExpression("value1"));
SpelExpressionParser parser = new SpelExpressionParser();
headers.put("header2", parser.parseExpression("context.timestamp() + ' @' + context.offset()"));
HeaderEnricher<String, String> enricher = new HeaderEnricher<>(headers);
KStream<String, String> stream = builder.stream(INPUT);
stream
        .transform(() -> enricher)
        .to(OUTPUT);

# 4.2.8. MessagingTransformer

版本 2.3 增加了 MessagingTransformer ，这允许 Kafka Streams 拓扑与 Spring 消息传递组件进行交互，例如 Spring 集成流。转换器要求实现 MessagingFunction 。

@FunctionalInterface
public interface MessagingFunction {
	Message<?> exchange(Message<?> message);

Spring 集成自动提供了一种使用其 GatewayProxyFactoryBean 的实现方式。它还需要一个 MessagingMessageConverter 来将键、值和元数据（包括头）转换为/来自 Spring 消息传递 Message<?> 。参见[[从 KStream 调用 Spring 集成流]（https://DOCS. Spring.io/ Spring-integration/DOCS/current/reference/html/kafka.html#Streams-integration）]以获得更多信息。

# 4.2.9.从反序列化异常恢复

版本 2.3 引入了 RecoveringDeserializationExceptionHandler ，它可以在发生反序列化异常时采取一些操作。请参考关于 DeserializationExceptionHandler 的 Kafka 文档，其中 RecoveringDeserializationExceptionHandler 是一个实现。 RecoveringDeserializationExceptionHandler 配置为 ConsumerRecordRecoverer 实现。该框架提供了 DeadLetterPublishingRecoverer ，它将失败的记录发送到死信主题。有关此回收器的更多信息，请参见 发布死信记录 。

要配置 recoverer，请将以下属性添加到你的 Streams 配置中：

@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
public KafkaStreamsConfiguration kStreamsConfigs() {
    Map<String, Object> props = new HashMap<>();
    props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
            RecoveringDeserializationExceptionHandler.class);
    props.put(RecoveringDeserializationExceptionHandler.KSTREAM_DESERIALIZATION_RECOVERER, recoverer());
    return new KafkaStreamsConfiguration(props);
@Bean
public DeadLetterPublishingRecoverer recoverer() {
    return new DeadLetterPublishingRecoverer(kafkaTemplate(),
            (record, ex) -> new TopicPartition("recovererDLQ", -1));

当然， recoverer() Bean 可以是你自己的 ConsumerRecordRecoverer 的实现。

# 4.2.10.Kafka Streams 示例

下面的示例结合了我们在本章中讨论的所有主题：

@Configuration
@EnableKafka
@EnableKafkaStreams
public static class KafkaStreamsConfig {
    @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
    public KafkaStreamsConfiguration kStreamsConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "testStreams");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass().getName());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        props.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, WallclockTimestampExtractor.class.getName());
        return new KafkaStreamsConfiguration(props);
    @Bean
    public StreamsBuilderFactoryBeanConfigurer configurer() {
        return fb -> fb.setStateListener((newState, oldState) -> {
            System.out.println("State transition from " + oldState + " to " + newState);
    @Bean
    public KStream<Integer, String> kStream(StreamsBuilder kStreamBuilder) {
        KStream<Integer, String> stream = kStreamBuilder.stream("streamingTopic1");
        stream
                .mapValues((ValueMapper<String, String>) String::toUpperCase)
                .groupByKey()
                .windowedBy(TimeWindows.of(Duration.ofMillis(1000)))
                .reduce((String value1, String value2) -> value1 + value2,
                		Named.as("windowStore"))
                .toStream()
                .map((windowedId, value) -> new KeyValue<>(windowedId.key(), value))
                .filter((i, s) -> s.length() > 40)
                .to("streamingTopic2");
        stream.print(Printed.toSysOut());
        return stream;

# 4.3.测试应用程序

spring-kafka-test JAR 包含一些有用的实用程序，以帮助测试你的应用程序。

# 4.3.1.Kafkatestutils

o.s.kafka.test.utils.KafkaTestUtils 提供了许多静态助手方法来使用记录、检索各种记录偏移量以及其他方法。有关完整的详细信息，请参阅其 Javadocs (opens new window) 。

# 4.3.2.朱尼特

o.s.kafka.test.utils.KafkaTestUtils 还提供了一些静态方法来设置生产者和消费者属性。下面的清单显示了这些方法签名：

/**
 * Set up test properties for an {@code <Integer, String>} consumer.
 * @param group the group id.
 * @param autoCommit the auto commit.
 * @param embeddedKafka a {@link EmbeddedKafkaBroker} instance.
 * @return the properties.
public static Map<String, Object> consumerProps(String group, String autoCommit,
                                       EmbeddedKafkaBroker embeddedKafka) { ... }
 * Set up test properties for an {@code <Integer, String>} producer.
 * @param embeddedKafka a {@link EmbeddedKafkaBroker} instance.
 * @return the properties.
public static Map<String, Object> producerProps(EmbeddedKafkaBroker embeddedKafka) { ... }

为 EmbeddedKafkaBroker 提供了一个 JUnit4 @Rule 包装器，用于创建嵌入式 Kafka 和嵌入式 ZooKeeper 服务器。（有关使用 @EmbeddedKafka 与 JUnit5 一起使用 @EmbeddedKafka 的信息，请参见 @Embeddedkafka 注释 ）。下面的清单显示了这些方法的签名：

/**
 * Create embedded Kafka brokers.
 * @param count the number of brokers.
 * @param controlledShutdown passed into TestUtils.createBrokerConfig.
 * @param topics the topics to create (2 partitions per).
public EmbeddedKafkaRule(int count, boolean controlledShutdown, String... topics) { ... }
 * Create embedded Kafka brokers.
 * @param count the number of brokers.
 * @param controlledShutdown passed into TestUtils.createBrokerConfig.
 * @param partitions partitions per topic.
 * @param topics the topics to create.
public EmbeddedKafkaRule(int count, boolean controlledShutdown, int partitions, String... topics) { ... }

EmbeddedKafkaBroker 类有一个实用程序方法，它允许你使用它创建的所有主题。下面的示例展示了如何使用它：

Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("testT", "false", embeddedKafka);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<Integer, String>(
        consumerProps);
Consumer<Integer, String> consumer = cf.createConsumer();
embeddedKafka.consumeFromAllEmbeddedTopics(consumer);

KafkaTestUtils 有一些实用方法来从使用者那里获取结果。下面的清单显示了这些方法签名：

/**
 * Poll the consumer, expecting a single record for the specified topic.
 * @param consumer the consumer.
 * @param topic the topic.
 * @return the record.
 * @throws org.junit.ComparisonFailure if exactly one record is not received.
public static <K, V> ConsumerRecord<K, V> getSingleRecord(Consumer<K, V> consumer, String topic) { ... }
 * Poll the consumer for records.
 * @param consumer the consumer.
 * @return the records.
public static <K, V> ConsumerRecords<K, V> getRecords(Consumer<K, V> consumer) { ... }

下面的示例展示了如何使用 KafkaTestUtils :

...
template.sendDefault(0, 2, "bar");
ConsumerRecord<Integer, String> received = KafkaTestUtils.getSingleRecord(consumer, "topic");

当 EmbeddedKafkaBroker 启动嵌入式 Kafka 和嵌入式 ZooKeeper 服务器时，将名为 spring.embedded.kafka.brokers 的系统属性设置为 Kafka 代理的地址，并将名为 spring.embedded.zookeeper.connect 的系统属性设置为 ZooKeeper 的地址。为此属性提供了方便的常量（ EmbeddedKafkaBroker.SPRING_EMBEDDED_KAFKA_BROKERS 和 EmbeddedKafkaBroker.SPRING_EMBEDDED_ZOOKEEPER_CONNECT ）。

使用 EmbeddedKafkaBroker.brokerProperties(Map<String, String>) ，你可以为 Kafka 服务器提供其他属性。有关可能的代理属性的更多信息，请参见 Kafka配置 (opens new window) 。

# 4.3.3.配置主题

下面的示例配置创建了带有五个分区的 cat 和 hat 主题，带有 10 个分区的 thing1 主题，以及带有 15 个分区的 thing2 主题：

public class MyTests {
    @ClassRule
    private static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, false, 5, "cat", "hat");
    @Test
    public void test() {
        embeddedKafkaRule.getEmbeddedKafka()
              .addTopics(new NewTopic("thing1", 10, (short) 1), new NewTopic("thing2", 15, (short) 1));

默认情况下， addTopics 在出现问题（例如添加已经存在的主题）时将抛出异常。版本 2.6 添加了该方法的新版本，该版本返回 Map<String, Exception> ；关键是主题名称，对于成功，值是 null ，对于失败，值是 Exception 。

# 4.3.4.对多个测试类使用相同的代理

这样做并没有内置的支持，但是你可以使用相同的代理对多个测试类进行类似于以下的操作：

public final class EmbeddedKafkaHolder {
    private static EmbeddedKafkaBroker embeddedKafka = new EmbeddedKafkaBroker(1, false)
            .brokerListProperty("spring.kafka.bootstrap-servers");
    private static boolean started;
    public static EmbeddedKafkaBroker getEmbeddedKafka() {
        if (!started) {
            try {
                embeddedKafka.afterPropertiesSet();
            catch (Exception e) {
                throw new KafkaException("Embedded broker failed to start", e);
            started = true;
        return embeddedKafka;
    private EmbeddedKafkaHolder() {
        super();

这假定启动环境为 Spring，并且嵌入式代理替换了 BootStrap Servers 属性。

然后，在每个测试类中，你可以使用类似于以下内容的内容：

static {
    EmbeddedKafkaHolder.getEmbeddedKafka().addTopics("topic1", "topic2");
private static final EmbeddedKafkaBroker broker = EmbeddedKafkaHolder.getEmbeddedKafka();

如果不使用 Spring boot，则可以使用 broker.getBrokersAsString() 获得 bootstrap 服务器。

# 4.3.5.@Embeddedkafka 注释

我们通常建议你使用 @ClassRule 规则，以避免在测试之间启动和停止代理（并为每个测试使用不同的主题）。从版本 2.0 开始，如果使用 Spring 的测试应用程序上下文缓存，还可以声明 EmbeddedKafkaBroker Bean，因此单个代理可以跨多个测试类使用。为了方便起见，我们提供了一个名为 @EmbeddedKafka 的测试类级注释来注册 EmbeddedKafkaBroker Bean。下面的示例展示了如何使用它：

@RunWith(SpringRunner.class)
@DirtiesContext
@EmbeddedKafka(partitions = 1,
         topics = {
                 KafkaStreamsTests.STREAMING_TOPIC1,
                 KafkaStreamsTests.STREAMING_TOPIC2 })
public class KafkaStreamsTests {
    @Autowired
    private EmbeddedKafkaBroker embeddedKafka;
    @Test
    public void someTest() {
        Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("testGroup", "true", this.embeddedKafka);
        consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        ConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
        Consumer<Integer, String> consumer = cf.createConsumer();
        this.embeddedKafka.consumeFromAnEmbeddedTopic(consumer, KafkaStreamsTests.STREAMING_TOPIC2);
        ConsumerRecords<Integer, String> replies = KafkaTestUtils.getRecords(consumer);
        assertThat(replies.count()).isGreaterThanOrEqualTo(1);
    @Configuration
    @EnableKafkaStreams
    public static class KafkaStreamsConfiguration {
        @Value("${" + EmbeddedKafkaBroker.SPRING_EMBEDDED_KAFKA_BROKERS + "}")
        private String brokerAddresses;
        @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
        public KafkaStreamsConfiguration kStreamsConfigs() {
            Map<String, Object> props = new HashMap<>();
            props.put(StreamsConfig.APPLICATION_ID_CONFIG, "testStreams");
            props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, this.brokerAddresses);
            return new KafkaStreamsConfiguration(props);

从版本 2.2.4 开始，你还可以使用 @EmbeddedKafka 注释来指定 Kafka Ports 属性。

下面的示例设置 topics 、 brokerProperties 和 brokerPropertiesLocation 属性的 @EmbeddedKafka 支持属性占位符解析：

@TestPropertySource(locations = "classpath:/test.properties")
@EmbeddedKafka(topics = { "any-topic", "${kafka.topics.another-topic}" },
        brokerProperties = { "log.dir=${kafka.broker.logs-dir}",
                            "listeners=PLAINTEXT://localhost:${kafka.broker.port}",
                            "auto.create.topics.enable=${kafka.broker.topics-enable:true}" },
        brokerPropertiesLocation = "classpath:/broker.properties")

在前面的示例中，属性占位符 ${kafka.topics.another-topic} 、 ${kafka.broker.logs-dir} 和 ${kafka.broker.port} 是从 Spring Environment 解析的。此外，代理属性是从 broker.properties Classpath 资源中加载的，该资源由 brokerPropertiesLocation 指定。属性占位符是为 brokerPropertiesLocation URL 和资源中找到的任何属性占位符解析的。由 brokerProperties 定义的属性覆盖在 brokerPropertiesLocation 中找到的属性。

你可以在 JUnit4 或 JUnit5 中使用 @EmbeddedKafka 注释。

# 4.3.6.@EmbeddedKafka 注释与 JUnit5

从版本 2.3 开始，有两种方法可以使用 JUnit5 的 @EmbeddedKafka 注释。当与 @SpringJunitConfig 注释一起使用时，嵌入式代理将添加到测试应用程序上下文中。你可以在类或方法级别将代理自动连接到你的测试中，以获得代理地址列表。

当 不是 使用 Spring 测试上下文时， EmbdeddedKafkaCondition 将创建代理；该条件包括一个参数解析程序，因此你可以在测试方法中访问代理…

@EmbeddedKafka
public class EmbeddedKafkaConditionTests {
    @Test
    public void test(EmbeddedKafkaBroker broker) {
        String brokerList = broker.getBrokersAsString();

如果用 ExtendedWith(SpringExtension.class) 注释的类也没有用 ExtendedWith(SpringExtension.class) 注释（或 meta 注释），则将创建一个独立的（而不是 Spring 测试上下文）代理。 @SpringJunitConfig 和 @SpringBootTest 是这样的元注释，并且基于上下文的代理将在也存在这些注释时使用。

# 4.3.7. @SpringBootTest 注释中的嵌入式代理

Spring Initializr (opens new window) 现在自动将测试范围中的 spring-kafka-test 依赖项添加到项目配置中。

在 Spring 引导应用程序测试中有几种使用嵌入式代理的方法。

它们包括：

• JUnit4 类规则

• [ @EmbeddedKafka 注释或 EmbeddedKafkaBroker Bean（#kafka-testing-embeddedkafka-annotation）

# JUnit4 类规则

下面的示例展示了如何使用 JUnit4 类规则来创建嵌入式代理：

@RunWith(SpringRunner.class)
@SpringBootTest
public class MyApplicationTests {
    @ClassRule
    public static EmbeddedKafkaRule broker = new EmbeddedKafkaRule(1,
        false, "someTopic")
            .brokerListProperty("spring.kafka.bootstrap-servers");
    @Autowired
    private KafkaTemplate<String, String> template;
    @Test
    public void test() {

注意，由于这是一个 Spring 引导应用程序，因此我们将覆盖代理列表属性以设置引导属性。

# @EmbeddedKafka 注释或 EmbeddedKafkaBroker Bean

下面的示例展示了如何使用 @EmbeddedKafka 注释来创建嵌入式代理：

@RunWith(SpringRunner.class)
@EmbeddedKafka(topics = "someTopic",
        bootstrapServersProperty = "spring.kafka.bootstrap-servers")
public class MyApplicationTests {
    @Autowired
    private KafkaTemplate<String, String> template;
    @Test
    public void test() {

# 4.3.8.汉克雷斯特火柴人

o.s.kafka.test.hamcrest.KafkaMatchers 提供了以下匹配器：

/**
 * @param key the key
 * @param <K> the type.
 * @return a Matcher that matches the key in a consumer record.
public static <K> Matcher<ConsumerRecord<K, ?>> hasKey(K key) { ... }
 * @param value the value.
 * @param <V> the type.
 * @return a Matcher that matches the value in a consumer record.
public static <V> Matcher<ConsumerRecord<?, V>> hasValue(V value) { ... }
 * @param partition the partition.
 * @return a Matcher that matches the partition in a consumer record.
public static Matcher<ConsumerRecord<?, ?>> hasPartition(int partition) { ... }
 * Matcher testing the timestamp of a {@link ConsumerRecord} assuming the topic has been set with
 * {@link org.apache.kafka.common.record.TimestampType#CREATE_TIME CreateTime}.
 * @param ts timestamp of the consumer record.
 * @return a Matcher that matches the timestamp in a consumer record.
public static Matcher<ConsumerRecord<?, ?>> hasTimestamp(long ts) {
  return hasTimestamp(TimestampType.CREATE_TIME, ts);
 * Matcher testing the timestamp of a {@link ConsumerRecord}
 * @param type timestamp type of the record
 * @param ts timestamp of the consumer record.
 * @return a Matcher that matches the timestamp in a consumer record.
public static Matcher<ConsumerRecord<?, ?>> hasTimestamp(TimestampType type, long ts) {
  return new ConsumerRecordTimestampMatcher(type, ts);

# 4.3.9.AssertJ 条件

你可以使用以下 AssertJ 条件：

/**
 * @param key the key
 * @param <K> the type.
 * @return a Condition that matches the key in a consumer record.
public static <K> Condition<ConsumerRecord<K, ?>> key(K key) { ... }
 * @param value the value.
 * @param <V> the type.
 * @return a Condition that matches the value in a consumer record.
public static <V> Condition<ConsumerRecord<?, V>> value(V value) { ... }
 * @param key the key.
 * @param value the value.
 * @param <K> the key type.
 * @param <V> the value type.
 * @return a Condition that matches the key in a consumer record.
 * @since 2.2.12
public static <K, V> Condition<ConsumerRecord<K, V>> keyValue(K key, V value) { ... }
 * @param partition the partition.
 * @return a Condition that matches the partition in a consumer record.
public static Condition<ConsumerRecord<?, ?>> partition(int partition) { ... }
 * @param value the timestamp.
 * @return a Condition that matches the timestamp value in a consumer record.
public static Condition<ConsumerRecord<?, ?>> timestamp(long value) {
  return new ConsumerRecordTimestampCondition(TimestampType.CREATE_TIME, value);
 * @param type the type of timestamp
 * @param value the timestamp.
 * @return a Condition that matches the timestamp value in a consumer record.
public static Condition<ConsumerRecord<?, ?>> timestamp(TimestampType type, long value) {
  return new ConsumerRecordTimestampCondition(type, value);

# 4.3.10.例子

下面的示例汇总了本章涵盖的大多数主题：

public class KafkaTemplateTests {
    private static final String TEMPLATE_TOPIC = "templateTopic";
    @ClassRule
    public static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, true, TEMPLATE_TOPIC);
    @Test
    public void testTemplate() throws Exception {
        Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("testT", "false",
            embeddedKafka.getEmbeddedKafka());
        DefaultKafkaConsumerFactory<Integer, String> cf =
                            new DefaultKafkaConsumerFactory<Integer, String>(consumerProps);
        ContainerProperties containerProperties = new ContainerProperties(TEMPLATE_TOPIC);
        KafkaMessageListenerContainer<Integer, String> container =
                            new KafkaMessageListenerContainer<>(cf, containerProperties);
        final BlockingQueue<ConsumerRecord<Integer, String>> records = new LinkedBlockingQueue<>();
        container.setupMessageListener(new MessageListener<Integer, String>() {
            @Override
            public void onMessage(ConsumerRecord<Integer, String> record) {
                System.out.println(record);
                records.add(record);
        container.setBeanName("templateTests");
        container.start();
        ContainerTestUtils.waitForAssignment(container,
                            embeddedKafka.getEmbeddedKafka().getPartitionsPerTopic());
        Map<String, Object> producerProps =
                            KafkaTestUtils.producerProps(embeddedKafka.getEmbeddedKafka());
        ProducerFactory<Integer, String> pf =
                            new DefaultKafkaProducerFactory<Integer, String>(producerProps);
        KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
        template.setDefaultTopic(TEMPLATE_TOPIC);
        template.sendDefault("foo");
        assertThat(records.poll(10, TimeUnit.SECONDS), hasValue("foo"));
        template.sendDefault(0, 2, "bar");
        ConsumerRecord<Integer, String> received = records.poll(10, TimeUnit.SECONDS);
        assertThat(received, hasKey(2));
        assertThat(received, hasPartition(0));
        assertThat(received, hasValue("bar"));
        template.send(TEMPLATE_TOPIC, 0, 2, "baz");
        received = records.poll(10, TimeUnit.SECONDS);
        assertThat(received, hasKey(2));
        assertThat(received, hasPartition(0));
        assertThat(received, hasValue("baz"));

前面的示例使用了 Hamcrest Matchers。使用 AssertJ ，最后一部分看起来像以下代码：

assertThat(records.poll(10, TimeUnit.SECONDS)).has(value("foo"));
template.sendDefault(0, 2, "bar");
ConsumerRecord<Integer, String> received = records.poll(10, TimeUnit.SECONDS);
// using individual assertions
assertThat(received).has(key(2));
assertThat(received).has(value("bar"));
assertThat(received).has(partition(0));
template.send(TEMPLATE_TOPIC, 0, 2, "baz");
received = records.poll(10, TimeUnit.SECONDS);
// using allOf()
assertThat(received).has(allOf(keyValue(2, "baz"), partition(0)));

# 4.4.非阻塞重试

使用 Kafka 实现非阻塞重试/DLT 功能通常需要设置额外的主题并创建和配置相应的侦听器。由于 2.7 Spring for Apache，Kafka 通过 @RetryableTopic 注释和 RetryTopicConfiguration 类提供了对此的支持，以简化该引导。

# 4.4.1.模式的工作原理

如果消息处理失败，该消息将被转发到带有后退时间戳的重试主题。然后，重试主题使用者检查时间戳，如果没有到期，它会暂停该主题分区的消耗。当它到期时，将恢复分区消耗，并再次消耗消息。如果消息处理再次失败，则消息将被转发到下一个重试主题，并重复该模式，直到处理成功，或者尝试已尽，并将消息发送到死信主题（如果已配置）。

为了说明这一点，如果你有一个“main-topic”主题，并且希望设置非阻塞重试，该重试的指数回退为 1000ms，乘数为 2 和 4max，那么它将创建 main-topic-retry-1000、main-topic-retry-2000、main-topic-retry-4000 和 main-topic-dlt 主题，并配置相应的消费者。该框架还负责创建主题以及设置和配置侦听器。

# 4.4.2.退后延迟精度

# 概述和保证

所有的消息处理和退线都由使用者线程处理，因此，在尽力而为的基础上保证了延迟精度。如果一条消息的处理时间超过了下一条消息对该消费者的回退期，则下一条消息的延迟将高于预期。此外，对于较短的延迟（大约 1s 或更短），线程必须进行的维护工作（例如提交偏移）可能会延迟消息处理的执行。如果重试主题的使用者正在处理多个分区，则精度也会受到影响，因为我们依赖于从轮询中唤醒使用者并具有完整的 polltimeouts 来进行时间调整。

话虽如此，对于处理单个分区的消费者来说，在大多数情况下，消息的处理时间应该在 100ms 以下。

# 调整延迟精度

消息的处理延迟精度依赖于两个 ContainerProperties : ContainerProperties.pollTimeout 和 ContainerProperties.idlePartitionEventInterval 。这两个属性将在重试主题和 DLT 的 ListenerContainerFactory 中自动设置为该主题最小延迟值的四分之一，最小值为 250ms，最大值为 5000ms。只有当属性有其默认值时，才会设置这些值-如果你自己更改其中一个值，你的更改将不会被重写。通过这种方式，你可以根据需要调整重试主题的精度和性能。

# 4.4.3.配置

# 使用 @RetryableTopic 注释

要为 @KafkaListener 注释方法配置重试主题和 DLT，只需向其添加 @RetryableTopic 注释，而 Spring 对于 Apache Kafka 将使用默认配置引导所有必要的主题和使用者。

@RetryableTopic(kafkaTemplate = "myRetryableTopicKafkaTemplate")
@KafkaListener(topics = "my-annotated-topic", groupId = "myGroupId")
public void processMessage(MyPojo message) {
        // ... message processing

你可以在同一个类中指定一个方法，通过使用 @DltHandler 注释来处理 DLT 消息。如果没有提供 Dlthandler 方法，则创建一个默认的使用者，该使用者只记录消费。

@DltHandler
public void processMessage(MyPojo message) {
// ... message processing, persistence, etc

# 使用 RetryTopicConfiguration bean

你还可以通过在 @Configuration 带注释的类中创建 RetryTopicConfiguration bean 来配置非阻塞重试支持。

@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, Object> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .create(template);

这将为使用默认配置以“@Kafkalistener”注释的方法中的所有主题创建重试主题和 DLT，以及相应的消费者。消息转发需要 KafkaTemplate 实例。

为了实现对如何处理每个主题的非阻塞重试的更细粒度的控制，可以提供一个以上的 RetryTopicConfiguration Bean。

@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, MyPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .fixedBackoff(3000)
            .maxAttempts(5)
            .includeTopics("my-topic", "my-other-topic")
            .create(template);
@Bean
public RetryTopicConfiguration myOtherRetryTopic(KafkaTemplate<String, MyOtherPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .exponentialBackoff(1000, 2, 5000)
            .maxAttempts(4)
            .excludeTopics("my-topic", "my-other-topic")
            .retryOn(MyException.class)
            .create(template);

@Bean
public ProducerFactory<String, Object> producerFactory() {
  return new DefaultKafkaProducerFactory<>(producerConfiguration(), new StringSerializer(),
    new DelegatingByTypeSerializer(Map.of(byte[].class, new ByteArraySerializer(),
          MyNormalObject.class, new JsonSerializer<Object>())));
@Bean
public KafkaTemplate<String, Object> kafkaTemplate() {
  return new KafkaTemplate<>(producerFactory());

# 4.4.4.特征

大多数特性都适用于 @RetryableTopic 注释和 RetryTopicConfiguration bean。

# 退避配置

退避配置依赖于 Spring Retry 项目中的 BackOffPolicy 接口。

它包括：

• 固定后退

• 指数式后退

• 随机指数回退

• 均匀随机退避

• 不退缩

• 自定义后退

@RetryableTopic(attempts = 5,
    backoff = @Backoff(delay = 1000, multiplier = 2, maxDelay = 5000))
@KafkaListener(topics = "my-annotated-topic")
public void processMessage(MyPojo message) {
        // ... message processing

@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, MyPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .fixedBackoff(3000)
            .maxAttempts(4)
            .build();

还可以提供 Spring Retry 的 SleepingBackOffPolicy 的自定义实现：

@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, MyPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .customBackOff(new MyCustomBackOffPolicy())
            .maxAttempts(5)
            .build();

# 单话题固定延迟重试

如果你使用固定的延迟策略，例如 FixedBackOffPolicy 或 NoBackOffPolicy ，你可以使用一个主题来完成非阻塞重试。此主题将使用提供的或默认的后缀作为后缀，并且不会附加索引或延迟值。

@RetryableTopic(backoff = @Backoff(2000), fixedDelayTopicStrategy = FixedDelayStrategy.SINGLE_TOPIC)
@KafkaListener(topics = "my-annotated-topic")
public void processMessage(MyPojo message) {
        // ... message processing

@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, MyPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .fixedBackoff(3000)
            .maxAttempts(5)
            .useSingleTopicForFixedDelays()
            .build();

# 全局超时

你可以为重试过程设置全局超时。如果达到了这个时间，则下一次使用者抛出异常时，消息将直接传递到 DLT，或者如果没有可用的 DLT，消息将结束处理。

@RetryableTopic(backoff = @Backoff(2000), timeout = 5000)
@KafkaListener(topics = "my-annotated-topic")
public void processMessage(MyPojo message) {
        // ... message processing

@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, MyPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .fixedBackoff(2000)
            .timeoutAfter(5000)
            .build();

# 异常分类器

你可以指定要重试的异常和不要重试的异常。你还可以将其设置为遍历原因以查找嵌套的异常。

@RetryableTopic(include = {MyRetryException.class, MyOtherRetryException.class}, traversingCauses = true)
@KafkaListener(topics = "my-annotated-topic")
public void processMessage(MyPojo message) {
        throw new RuntimeException(new MyRetryException()); // Will retry

@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, MyOtherPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .notRetryOn(MyDontRetryException.class)
            .create(template);

从 2.8.3 开始，有一个致命异常的全局列表，它将导致记录在没有任何重试的情况下被发送到 DLT。有关致命异常的默认列表，请参见 违约恐怖处理者 。你可以通过以下方式向该列表添加或删除异常：

@Bean(name = RetryTopicInternalBeanNames.DESTINATION_TOPIC_CONTAINER_NAME)
public DefaultDestinationTopicResolver topicResolver(ApplicationContext applicationContext,
                                               @Qualifier(RetryTopicInternalBeanNames
                                                       .INTERNAL_BACKOFF_CLOCK_BEAN_NAME) Clock clock) {
    DefaultDestinationTopicResolver ddtr = new DefaultDestinationTopicResolver(clock, applicationContext);
    ddtr.addNotRetryableExceptions(MyFatalException.class);
    ddtr.removeNotRetryableException(ConversionException.class);
    return ddtr;

# 包含和排除主题

你可以通过.includeTopic（字符串主题）、.includeTopics（集合 <gtr="2886"/>主题）、.excludeTopic（字符串主题）和.excludeTopics（集合 <gtr="2887"/>主题）方法来决定哪些主题将由<gtr="2885"/> Bean 处理。

@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<Integer, MyPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .includeTopics(List.of("my-included-topic", "my-other-included-topic"))
            .create(template);
@Bean
public RetryTopicConfiguration myOtherRetryTopic(KafkaTemplate<Integer, MyPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .excludeTopic("my-excluded-topic")
            .create(template);

# topics 自动创建

除非另有说明，否则框架将使用 NewTopic bean 自动创建所需的主题，这些 bean 由 KafkaAdmin Bean 使用。你可以指定创建主题所使用的分区数量和复制因子，并且可以关闭此功能。

@RetryableTopic(numPartitions = 2, replicationFactor = 3)
@KafkaListener(topics = "my-annotated-topic")
public void processMessage(MyPojo message) {
        // ... message processing
@RetryableTopic(autoCreateTopics = false)
@KafkaListener(topics = "my-annotated-topic")
public void processMessage(MyPojo message) {
        // ... message processing

@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<Integer, MyPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .autoCreateTopicsWith(2, 3)
            .create(template);
@Bean
public RetryTopicConfiguration myOtherRetryTopic(KafkaTemplate<Integer, MyPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .doNotAutoCreateRetryTopics()
            .create(template);

# 故障报头管理

在考虑如何管理故障报头（原始报头和异常报头）时，框架将委托给 DeadLetterPublishingRecover ，以决定是否追加或替换报头。

默认情况下，它显式地将 appendOriginalHeaders 设置为 false ，并将 stripPreviousExceptionHeaders 设置为 DeadLetterPublishingRecover 使用的默认值。

这意味着默认配置只保留第一个“原始”和最后一个异常标头。这是为了避免在涉及许多重试步骤时创建过大的消息（例如，由于堆栈跟踪标头）。

有关更多信息，请参见 管理死信记录头 。

要重新配置框架以对这些属性使用不同的设置，请通过添加 recovererCustomizer 来替换标准 DeadLetterPublishingRecovererFactory Bean:

@Bean(RetryTopicInternalBeanNames.DEAD_LETTER_PUBLISHING_RECOVERER_FACTORY_BEAN_NAME)
DeadLetterPublishingRecovererFactory factory(DestinationTopicResolver resolver) {
    DeadLetterPublishingRecovererFactory factory = new DeadLetterPublishingRecovererFactory(resolver);
    factory.setDeadLetterPublishingRecovererCustomizer(dlpr -> {
        dlpr.appendOriginalHeaders(true);
        dlpr.setStripPreviousExceptionHeaders(false);
    return factory;

# 4.4.5.主题命名

Retry Topics 和 DLT 的命名方法是使用提供的或默认值对主主题进行后缀，并附加该主题的延迟或索引。

例子：

“my-topic”“my-topic-retry-0”，“my-topic-retry-1”，…，“my-topic-dlt”

“my-other-topic”“my-topic-myretrySuffix-1000”，“my-topic-myretrySuffix-2000”，…，“my-topic-mydltSufix”。

# 重试主题和 DLT 后缀

你可以指定 Retry 和 DLT 主题将使用的后缀。

@RetryableTopic(retryTopicSuffix = "-my-retry-suffix", dltTopicSuffix = "-my-dlt-suffix")
@KafkaListener(topics = "my-annotated-topic")
public void processMessage(MyPojo message) {
        // ... message processing

@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, MyOtherPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .retryTopicSuffix("-my-retry-suffix")
            .dltTopicSuffix("-my-dlt-suffix")
            .create(template);

# 附加主题索引或延迟

你可以在后缀之后追加主题的索引值，也可以在后缀之后追加延迟值。

@RetryableTopic(topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE)
@KafkaListener(topics = "my-annotated-topic")
public void processMessage(MyPojo message) {
        // ... message processing

@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, MyPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .suffixTopicsWithIndexValues()
            .create(template);

# 自定义命名策略

可以通过注册实现 RetryTopicNamesProviderFactory 的 Bean 来实现更复杂的命名策略。默认实现是 SuffixingRetryTopicNamesProviderFactory ，可以通过以下方式注册不同的实现：

@Bean
public RetryTopicNamesProviderFactory myRetryNamingProviderFactory() {
    return new CustomRetryTopicNamesProviderFactory();

作为示例，下面的实现除了标准后缀之外，还添加了一个前缀来 Retry/DL 主题名称：

public class CustomRetryTopicNamesProviderFactory implements RetryTopicNamesProviderFactory {
	@Override
    public RetryTopicNamesProvider createRetryTopicNamesProvider(
                DestinationTopic.Properties properties) {
        if(properties.isMainEndpoint()) {
            return new SuffixingRetryTopicNamesProvider(properties);
        else {
            return new SuffixingRetryTopicNamesProvider(properties) {
                @Override
                public String getTopicName(String topic) {
                    return "my-prefix-" + super.getTopicName(topic);

# 4.4.6.DLT 策略

该框架为使用 DLTS 提供了一些策略。你可以提供用于 DLT 处理的方法，也可以使用默认的日志记录方法，或者根本没有 DLT。你还可以选择如果 DLT 处理失败会发生什么。

# DLT 处理方法

你可以指定用于处理该主题的 DLT 的方法，以及在处理失败时的行为。

要做到这一点，你可以在具有 @RetryableTopic 注释的类的方法中使用 @DltHandler 注释。请注意，相同的方法将用于该类中的所有 @RetryableTopic 注释方法。

@RetryableTopic
@KafkaListener(topics = "my-annotated-topic")
public void processMessage(MyPojo message) {
        // ... message processing
@DltHandler
public void processMessage(MyPojo message) {
// ... message processing, persistence, etc

DLT 处理程序方法也可以通过 RetryTopicConfigurationBuilder.dlthandlerMethod 方法提供，将处理 DLT 消息的 Bean 名称和方法名称作为参数传递。

@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<Integer, MyPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .dltProcessor("myCustomDltProcessor", "processDltMessage")
            .create(template);
@Component
public class MyCustomDltProcessor {
    private final MyDependency myDependency;
    public MyCustomDltProcessor(MyDependency myDependency) {
        this.myDependency = myDependency;
    public void processDltMessage(MyPojo message) {
       // ... message processing, persistence, etc

从版本 2.8 开始，如果你根本不想在此应用程序中使用 DLT，包括通过默认处理程序（或者你希望延迟使用），则可以控制 DLT 容器是否开始，这与容器工厂的 autoStartup 属性无关。

当使用 @RetryableTopic 注释时，将 autoStartDltHandler 属性设置为 false ；当使用配置生成器时，使用 .autoStartDltHandler(false) 。

稍后可以通过 KafkaListenerEndpointRegistry 启动 DLT 处理程序。

# DLT 故障行为

如果 DLT 处理失败，有两种可能的行为可用： ALWAYS_RETRY_ON_ERROR 和 FAIL_ON_ERROR 。

在前者中，记录被转发回 DLT 主题，因此它不会阻止其他 DLT 记录的处理。在后一种情况下，使用者在不转发消息的情况下结束执行。

@RetryableTopic(dltProcessingFailureStrategy =
			DltStrategy.FAIL_ON_ERROR)
@KafkaListener(topics = "my-annotated-topic")
public void processMessage(MyPojo message) {
        // ... message processing

@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<Integer, MyPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .dltProcessor(MyCustomDltProcessor.class, "processDltMessage")
            .doNotRetryOnDltFailure()
            .create(template);

被认为是致命的例外是：

• DeserializationException

• MessageConversionException

• ConversionException

• MethodArgumentResolutionException

• NoSuchMethodException

• ClassCastException

可以使用 DestinationTopicResolver Bean 上的方法向该列表添加异常并从该列表中删除异常。

有关更多信息，请参见 异常分类器 。

# 配置无 dlt

该框架还提供了不为主题配置 DLT 的可能性。在这种情况下，在重审用尽之后，程序就结束了。

@RetryableTopic(dltProcessingFailureStrategy =
			DltStrategy.NO_DLT)
@KafkaListener(topics = "my-annotated-topic")
public void processMessage(MyPojo message) {
        // ... message processing

@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<Integer, MyPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .doNotConfigureDlt()
            .create(template);

# 4.4.7.指定 ListenerContainerFactory

默认情况下，RetryTopic 配置将使用 @KafkaListener 注释中提供的工厂，但是你可以指定一个不同的工厂来创建 RetryTopic 和 DLT 侦听器容器。

对于 @RetryableTopic 注释，你可以提供工厂的 Bean 名称，并且使用 RetryTopicConfiguration Bean 你可以提供 Bean 名称或实例本身。

@RetryableTopic(listenerContainerFactory = "my-retry-topic-factory")
@KafkaListener(topics = "my-annotated-topic")
public void processMessage(MyPojo message) {
        // ... message processing

@Bean
public RetryTopicConfiguration myRetryTopic(KafkaTemplate<Integer, MyPojo> template,
        ConcurrentKafkaListenerContainerFactory<Integer, MyPojo> factory) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .listenerFactory(factory)
            .create(template);
@Bean
public RetryTopicConfiguration myOtherRetryTopic(KafkaTemplate<Integer, MyPojo> template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .listenerFactory("my-retry-topic-factory")
            .create(template);

如果需要将工厂配置行为恢复到 Prior2.8.3，则可以替换标准的 RetryTopicConfigurer Bean，并将 useLegacyFactoryConfigurer 设置为 true ，例如：

@Bean(name = RetryTopicInternalBeanNames.RETRY_TOPIC_CONFIGURER)
public RetryTopicConfigurer retryTopicConfigurer(DestinationTopicProcessor destinationTopicProcessor,
                                                ListenerContainerFactoryResolver containerFactoryResolver,
                                                ListenerContainerFactoryConfigurer listenerContainerFactoryConfigurer,
                                                BeanFactory beanFactory,
                                                RetryTopicNamesProviderFactory retryTopicNamesProviderFactory) {
    RetryTopicConfigurer retryTopicConfigurer = new RetryTopicConfigurer(destinationTopicProcessor, containerFactoryResolver, listenerContainerFactoryConfigurer, beanFactory, retryTopicNamesProviderFactory);
    retryTopicConfigurer.useLegacyFactoryConfigurer(true);
    return retryTopicConfigurer;

==== 更改 KafkabackoffException 日志记录级别

当重试主题中的消息未到期消耗时，将抛出 KafkaBackOffException 。默认情况下，这种异常会在 DEBUG 级别记录，但是你可以通过在 ListenerContainerFactoryConfigurer 类中的 @Configuration 中设置错误处理程序自定义程序来更改此行为。

例如，要更改日志级别以发出警告，你可以添加：

@Bean(name = RetryTopicInternalBeanNames.LISTENER_CONTAINER_FACTORY_CONFIGURER_NAME)
public ListenerContainerFactoryConfigurer listenerContainer(KafkaConsumerBackoffManager kafkaConsumerBackoffManager,
                                                            DeadLetterPublishingRecovererFactory deadLetterPublishingRecovererFactory,
                                                            @Qualifier(RetryTopicInternalBeanNames
                                                                    .INTERNAL_BACKOFF_CLOCK_BEAN_NAME) Clock clock) {
    ListenerContainerFactoryConfigurer configurer = new ListenerContainerFactoryConfigurer(kafkaConsumerBackoffManager, deadLetterPublishingRecovererFactory, clock);
    configurer.setErrorHandlerCustomizer(commonErrorHandler -> ((DefaultErrorHandler) commonErrorHandler).setLogLevel(KafkaException.Level.WARN));
    return configurer;

== 提示、技巧和示例

=== 手动分配所有分区

假设你总是希望读取所有分区的所有记录（例如，当使用压缩主题加载分布式缓存时），手动分配分区而不使用 Kafka 的组管理可能会很有用。当有许多分区时，这样做可能会很麻烦，因为你必须列出分区。如果分区的数量随着时间的推移而变化，这也是一个问题，因为每次分区数量发生变化时，你都必须重新编译应用程序。

下面是一个示例，说明如何在应用程序启动时使用 SPEL 表达式的能力动态地创建分区列表：

@KafkaListener(topicPartitions = @TopicPartition(topic = "compacted",
            partitions = "#{@finder.partitions('compacted')}"),
            partitionOffsets = @PartitionOffset(partition = "*", initialOffset = "0")))
public void listen(@Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String key, String payload) {
@Bean
public PartitionFinder finder(ConsumerFactory<String, String> consumerFactory) {
    return new PartitionFinder(consumerFactory);
public static class PartitionFinder {
    private final ConsumerFactory<String, String> consumerFactory;
    public PartitionFinder(ConsumerFactory<String, String> consumerFactory) {
        this.consumerFactory = consumerFactory;
    public String[] partitions(String topic) {
        try (Consumer<String, String> consumer = consumerFactory.createConsumer()) {
            return consumer.partitionsFor(topic).stream()
                .map(pi -> "" + pi.partition())
                .toArray(String[]::new);

将此与 ConsumerConfig.AUTO_OFFSET_RESET_CONFIG=earliest 结合使用，将在每次启动应用程序时加载所有记录。你还应该将容器的 AckMode 设置为 MANUAL ，以防止容器提交 null 消费者组的偏移。然而，从版本 2.5.5 开始，如上面所示，你可以对所有分区应用初始偏移量；有关更多信息，请参见 显式分区分配 。

=== 与其他事务管理器的 Kafka 事务示例

Spring 下面的引导应用程序是链接数据库和 Kafka 事务的一个示例。侦听器容器启动 Kafka 事务， @Transactional 注释启动 DB 事务。首先提交 DB 事务；如果 Kafka 事务未能提交，则将重新交付记录，因此 DB 更新应该是幂等的。

@SpringBootApplication
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    @Bean
    public ApplicationRunner runner(KafkaTemplate<String, String> template) {
        return args -> template.executeInTransaction(t -> t.send("topic1", "test"));
    @Bean
    public DataSourceTransactionManager dstm(DataSource dataSource) {
        return new DataSourceTransactionManager(dataSource);
    @Component
    public static class Listener {
        private final JdbcTemplate jdbcTemplate;
        private final KafkaTemplate<String, String> kafkaTemplate;
        public Listener(JdbcTemplate jdbcTemplate, KafkaTemplate<String, String> kafkaTemplate) {
            this.jdbcTemplate = jdbcTemplate;
            this.kafkaTemplate = kafkaTemplate;
        @KafkaListener(id = "group1", topics = "topic1")
        @Transactional("dstm")
        public void listen1(String in) {
            this.kafkaTemplate.send("topic2", in.toUpperCase());
            this.jdbcTemplate.execute("insert into mytable (data) values ('" + in + "')");
        @KafkaListener(id = "group2", topics = "topic2")
        public void listen2(String in) {
            System.out.println(in);
    @Bean
    public NewTopic topic1() {
        return TopicBuilder.name("topic1").build();
    @Bean
    public NewTopic topic2() {
        return TopicBuilder.name("topic2").build();

spring.datasource.url=jdbc:mysql://localhost/integration?serverTimezone=UTC
spring.datasource.username=root
spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.enable-auto-commit=false
spring.kafka.consumer.properties.isolation.level=read_committed
spring.kafka.producer.transaction-id-prefix=tx-
#logging.level.org.springframework.transaction=trace
#logging.level.org.springframework.kafka.transaction=debug
#logging.level.org.springframework.jdbc=debug

create table mytable (data varchar(20));

对于仅用于生产者的事务，事务同步工作：

@Transactional("dstm")
public void someMethod(String in) {
    this.kafkaTemplate.send("topic2", in.toUpperCase());
    this.jdbcTemplate.execute("insert into mytable (data) values ('" + in + "')");

KafkaTemplate 将使其事务与 DB 事务同步，并且在数据库之后发生提交/回滚。

如果你希望首先提交 Kafka 事务，并且仅在 Kafka 事务成功的情况下提交 DB 事务，请使用嵌套 @Transactional 方法：

@Transactional("dstm")
public void someMethod(String in) {
    this.jdbcTemplate.execute("insert into mytable (data) values ('" + in + "')");
    sendToKafka(in);
@Transactional("kafkaTransactionManager")
public void sendToKafka(String in) {
    this.kafkaTemplate.send("topic2", in.toUpperCase());

=== 定制 JSONSerializer 和 JSONDESerializer

序列化器和反序列化器支持使用属性进行许多 cusomization，有关更多信息，请参见 JSON 。将这些对象实例化的代码（而不是 Spring）是 kafka-clients 代码，除非你将它们直接注入到消费者工厂和生产者工厂。如果你希望使用属性配置（去）序列化器，但是希望使用自定义 ObjectMapper ，那么只需创建一个子类并将自定义映射器传递到 super 构造函数中。例如：

public class CustomJsonSerializer extends JsonSerializer<Object> {
    public CustomJsonSerializer() {
        super(customizedObjectMapper());
    private static ObjectMapper customizedObjectMapper() {
        ObjectMapper mapper = JacksonUtils.enhancedObjectMapper();
        mapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
        return mapper;

== 其他资源

除了这个参考文档，我们还推荐了许多其他资源，这些资源可能有助于你了解 Spring 和 Apache Kafka。

• Apache Kafka Project Home Page (opens new window)

• Spring for Apache Kafka Home Page (opens new window)

• Spring for Apache Kafka GitHub Repository (opens new window)

• Spring Integration GitHub Repository (Apache Kafka Module) (opens new window)

== 覆盖 Spring 引导依赖项

当在 Spring 引导应用程序中对 Apache Kafka 使用 Spring 时， Apache Kafka 依赖关系版本由 Spring 引导的依赖关系管理确定。如果希望使用不同版本的 kafka-clients 或 kafka-streams ，并使用嵌入式 Kafka 代理进行测试，则需要覆盖 Spring 引导依赖项管理使用的版本，并为 Apache Kafka 添加两个 test 工件。

Maven

<properties>
    <kafka.version>2.8.1</kafka.version>
</properties>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>
<!-- optional - only needed when using kafka-streams -->
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka-test</artifactId>
    <scope>test</scope>
     <!-- needed if downgrading to Apache Kafka 2.8.1 -->
    <exclusions>
        <exclusion>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
        </exclusion>
    </exclusions>
</dependency>
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <classifier>test</classifier>
    <scope>test</scope>
    <version>${kafka.version}</version>
</dependency>
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka_2.13</artifactId>
    <classifier>test</classifier>
    <scope>test</scope>
    <version>${kafka.version}</version>
</dependency>

Gradle

ext['kafka.version'] = '2.8.1'
dependencies {
    implementation 'org.springframework.kafka:spring-kafka'
    implementation "org.apache.kafka:kafka-streams" // optional - only needed when using kafka-streams
    testImplementation ('org.springframework.kafka:spring-kafka-test') {
            // needed if downgrading to Apache Kafka 2.8.1
            exclude group: 'org.apache.zookeeper', module: 'zookeeper'
    testImplementation "org.apache.kafka:kafka-clients:${kafka.version}:test"