版本：5.0.1

本站内所有文档均为中英对照文档，点击中文可以显示英文。此提醒连续关闭5次后，将不再显示。

文档内容来源于 spring.io，由 springdoc.tech 翻译，版权归属于 SPRING.IO (Broadcom. Inc)。可供个人学习、研究，未经许可，不得进行转载或用于商业行为。

技巧、窍门与实用方法

DeepSeek V3 中英对照 Tips, Tricks and Recipes

使用 Kafka 的简单 DLQ

问题陈述

作为一名开发者，我想编写一个消费者应用程序来处理来自Kafka主题的记录。然而，如果在处理过程中出现错误，我不希望应用程序完全停止。相反，我希望将出错的记录发送到DLT（死信主题），然后继续处理新记录。

解决方案

这个问题的解决方案是使用 Spring Cloud Stream 中的 DLQ 功能。为了便于讨论，我们假设以下是我们使用的处理器函数。

@Bean
public Consumer<byte[]> processData() {
  return s -> {
     throw new RuntimeException();
  };
}

这是一个非常简单的函数，它会为所有处理的记录抛出异常，但你可以借鉴这个函数并将其扩展到任何其他类似的情况。

为了将错误记录发送到DLT，我们需要提供以下配置。

spring.cloud.stream:
  bindings:
   processData-in-0:
     group: my-group
     destination: input-topic
 kafka:
   bindings:
     processData-in-0:
       consumer:
         enableDlq: true
         dlqName: input-topic-dlq

要激活死信队列（DLQ），应用程序必须提供一个消费者组名称。匿名消费者无法使用死信队列功能。我们还需要在Kafka消费者绑定上设置 enableDLQ 属性为 true 来启用死信队列。最后，我们可以选择性地通过在Kafka消费者绑定上提供 dlqName 来指定死信主题（DLT）的名称，否则在这种情况下，默认名称将是 error.input-topic.my-group。

请注意，在上方提供的示例消费者中，payload 的类型为 byte[]。默认情况下，Kafka binder 中的 DLQ 生产者期望 payload 的类型为 byte[]。如果不是这种情况，那么我们需要提供正确的序列化器配置。例如，让我们将消费者函数重写如下：

@Bean
public Consumer<String> processData() {
  return s -> {
     throw new RuntimeException();
  };
}

现在，我们需要告诉 Spring Cloud Stream，在写入 DLT 时我们希望如何序列化数据。以下是针对此场景修改后的配置：

spring.cloud.stream:
  bindings:
   processData-in-0:
     group: my-group
     destination: input-topic
 kafka:
   bindings:
     processData-in-0:
       consumer:
         enableDlq: true
         dlqName: input-topic-dlq
         dlqProducerProperties:
           configuration:
             value.serializer: org.apache.kafka.common.serialization.StringSerializer

具有高级重试选项的 DLQ

问题陈述

这与上面的方法类似，但作为开发者，我希望能够配置重试的处理方式。

解决方案

如果你遵循了上述配置，那么当处理过程中遇到错误时，你将获得 Kafka binder 内置的默认重试选项。

默认情况下，绑定器最多重试 3 次，初始延迟为 1 秒，每次退避的乘数为 2.0，最大延迟为 10 秒。你可以按如下方式更改所有这些配置：

spring.cloud.stream.bindings.processData-in-0.consumer.maxAttempts
spring.cloud.stream.bindings.processData-in-0.consumer.backOffInitialInterval
spring.cloud.stream.bindings.processData-in-0.consumer.backOffMultiplier
spring.cloud.stream.bindings.processData-in-0.consumer.backOffMaxInterval

如果你愿意，也可以通过提供一个布尔值映射来指定可重试的异常列表。例如，

spring.cloud.stream.bindings.processData-in-0.consumer.retryableExceptions.java.lang.IllegalStateException=true
spring.cloud.stream.bindings.processData-in-0.consumer.retryableExceptions.java.lang.IllegalArgumentException=false

默认情况下，上表中未列出的任何异常都将被重试。如果不需要此行为，可以通过提供以下配置来禁用：

spring.cloud.stream.bindings.processData-in-0.consumer.defaultRetryable=false

你也可以提供自己的 RetryTemplate 并将其标记为 @StreamRetryTemplate，这样绑定器就会扫描并使用它。当你需要更复杂的重试策略和策略时，这非常有用。

如果你有多个 @StreamRetryTemplate bean，那么你可以通过使用属性来指定你的绑定希望使用哪一个，

spring.cloud.stream.bindings.processData-in-0.consumer.retry-template-name=<your-retry-template-bean-name>

使用 DLQ 处理反序列化错误

问题陈述

我的处理器在Kafka消费者中遇到了反序列化异常。我原本期望Spring Cloud Stream的DLQ机制能够捕获这种情况，但它并没有。我该如何处理这个问题？

解决方案

当 Kafka 消费者抛出不可恢复的反序列化异常时，Spring Cloud Stream 提供的常规 DLQ 机制将无法提供帮助。这是因为，这种异常甚至在消费者的 poll() 方法返回之前就已经发生。Spring for Apache Kafka 项目提供了一些很好的方法来帮助 binder 处理这种情况。让我们来探讨一下这些方法。

假设这是我们的函数：

@Bean
public Consumer<String> functionName() {
    return s -> {
        System.out.println(s);
    };
}

这是一个简单的函数，它接受一个 String 参数。

我们希望绕过 Spring Cloud Stream 提供的消息转换器，转而使用原生反序列化器。对于 String 类型来说，这并没有太大意义，但对于更复杂的类型（如 AVRO 等），您必须依赖外部反序列化器，因此希望将转换委托给 Kafka。

现在，当消费者接收到数据时，假设有一条不良记录导致了反序列化错误，例如，有人传递了一个 Integer 而不是 String。在这种情况下，如果你不在应用程序中采取任何措施，异常将通过调用链传播，最终导致你的应用程序退出。

为了处理这种情况，您可以添加一个 ListenerContainerCustomizer @Bean 来配置 DefaultErrorHandler。这个 DefaultErrorHandler 配置了一个 DeadLetterPublishingRecoverer。我们还需要为消费者配置一个 ErrorHandlingDeserializer。这听起来很复杂，但实际上，在这种情况下，它归结为这 3 个 bean。

@Bean
	public ListenerContainerCustomizer<AbstractMessageListenerContainer<byte[], byte[]>> customizer(DefaultErrorHandler errorHandler) {
		return (container, dest, group) -> {
			container.setCommonErrorHandler(errorHandler);
		};
	}

@Bean
	public DefaultErrorHandler errorHandler(DeadLetterPublishingRecoverer deadLetterPublishingRecoverer) {
		return new DefaultErrorHandler(deadLetterPublishingRecoverer);
	}

@Bean
	public DeadLetterPublishingRecoverer publisher(KafkaOperations bytesTemplate) {
		return new DeadLetterPublishingRecoverer(bytesTemplate);
	}

让我们逐一分析它们。第一个是带有 DefaultErrorHandler 的 ListenerContainerCustomizer bean。现在容器已使用该特定错误处理程序进行了自定义。您可以在此处了解更多关于容器自定义的信息。

第二个bean是配置为发布到DLT的DefaultErrorHandler。有关DefaultErrorHandler的更多详细信息，请参见此处。

第三个bean是DeadLetterPublishingRecoverer，它最终负责将消息发送到DLT。默认情况下，DLT主题被命名为ORIGINAL_TOPIC_NAME.DLT。不过，你可以更改这个命名。更多详情请参阅文档。

我们还需要通过应用配置来配置一个 ErrorHandlingDeserializer。

ErrorHandlingDeserializer 委托给实际的解串器。当发生错误时，它会将记录的键/值设置为 null，并包含消息的原始字节。然后，它在标头中设置异常，并将此记录传递给监听器，监听器随后调用已注册的错误处理程序。

以下是所需的配置：

spring.cloud.stream:
  function:
    definition: functionName
  bindings:
    functionName-in-0:
      group: group-name
      destination: input-topic
      consumer:
       use-native-decoding: true
  kafka:
    bindings:
      functionName-in-0:
        consumer:
          enableDlq: true
          dlqName: dlq-topic
          dlqProducerProperties:
            configuration:
              value.serializer: org.apache.kafka.common.serialization.StringSerializer
          configuration:
            value.deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
            spring.deserializer.value.delegate.class: org.apache.kafka.common.serialization.StringDeserializer

我们通过绑定上的 configuration 属性提供 ErrorHandlingDeserializer。同时，我们指定要委托的实际反序列化器是 StringDeserializer。

请注意，以上讨论的 dlq 属性与本方案中的讨论无关。它们仅用于处理应用程序级别的错误。

Kafka绑定器中的基本偏移量管理

问题陈述

我想编写一个Spring Cloud Stream Kafka消费者应用程序，但不太清楚它如何管理Kafka消费者偏移量。你能解释一下吗？

解决方案

我们建议您阅读相关文档章节，以全面理解相关内容。

简单来说：

Kafka默认支持两种起始偏移量类型——earliest和latest。从名称即可直观理解其语义。

假设您是首次运行消费者。如果在您的Spring Cloud Stream应用程序中未指定group.id，那么它将成为匿名消费者。每当您有一个匿名消费者时，Spring Cloud Stream应用程序默认会从主题分区中最新可用的偏移量开始消费。另一方面，如果您明确指定了group.id，那么默认情况下，Spring Cloud Stream应用程序将从主题分区中最早可用的偏移量开始消费。

在上述两种情况下（具有显式分组的消费者和匿名分组），可以通过使用属性 spring.cloud.stream.kafka.bindings.<binding-name>.consumer.startOffset 并将其设置为 earliest 或 latest 来调整起始偏移量。

现在，假设您之前已经运行过消费者，现在再次启动它。在这种情况下，上述情况中的起始偏移量语义不再适用，因为消费者会找到消费者组已提交的偏移量（对于匿名消费者，尽管应用程序未提供 group.id，绑定器会自动为您生成一个）。消费者会直接从最后提交的偏移量开始消费。即使您提供了 startOffset 值，这一点仍然成立。

然而，你可以通过使用 resetOffsets 属性来覆盖默认行为，即消费者从最后提交的偏移量开始消费。为此，将属性 spring.cloud.stream.kafka.bindings.<binding-name>.consumer.resetOffsets 设置为 true（默认为 false）。然后确保提供 startOffset 值（可以是 earliest 或 latest）。当你这样做并启动消费者应用程序时，每次启动时，它都会像第一次启动一样开始，并忽略分区中任何已提交的偏移量。

在 Kafka 中定位到任意偏移量

问题陈述

使用Kafka绑定器时，我知道可以将偏移量设置为earliest或latest，但我有一个需求，需要将偏移量定位到中间某个位置，即任意偏移量。有没有办法通过Spring Cloud Stream Kafka绑定器实现这一点？

解决方案

之前我们已经了解了Kafka绑定器如何处理基本的偏移量管理。默认情况下，绑定器不允许您回退到任意偏移量，至少不能通过我们在那个配方中看到的机制来实现。然而，绑定器提供了一些底层策略来实现这个用例。让我们来探索一下。

首先，当您想要重置到除 earliest 或 latest 之外的任意偏移量时，请确保将 resetOffsets 配置保留为其默认值，即 false。然后，您需要提供一个类型为 KafkaBindingRebalanceListener 的自定义 bean，该 bean 将被注入到所有消费者绑定中。这是一个带有一些默认方法的接口，但这里是我们感兴趣的方法：

/**
	 * Invoked when partitions are initially assigned or after a rebalance. Applications
	 * might only want to perform seek operations on an initial assignment. While the
	 * 'initial' argument is true for each thread (when concurrency is greater than 1),
	 * implementations should keep track of exactly which partitions have been sought.
	 * There is a race in that a rebalance could occur during startup and so a topic/
	 * partition that has been sought on one thread may be re-assigned to another
	 * thread and you may not wish to re-seek it at that time.
	 * @param bindingName the name of the binding.
	 * @param consumer the consumer.
	 * @param partitions the partitions.
	 * @param initial true if this is the initial assignment on the current thread.
	 */
	default void onPartitionsAssigned(String bindingName, Consumer<?, ?> consumer,
			Collection<TopicPartition> partitions, boolean initial) {
		// do nothing
	}

让我们来看一下细节。

本质上，这个方法会在每次主题分区初始分配或重新平衡后被调用。为了更好地说明，假设我们的主题是 foo，它有4个分区。最初，我们只在消费者组中启动一个消费者，这个消费者将消费所有分区。当消费者首次启动时，所有4个分区都会被初始分配。然而，我们并不希望分区从默认位置开始消费（因为我们定义了一个消费者组，所以默认是 earliest），而是希望每个分区在定位到任意偏移量后再开始消费。想象一下，你有一个业务场景需要从如下所示的特定偏移量开始消费。

Partition   start offset

         1000
         2000
         2000
         1000

这可以通过如下实现上述方法来实现。

@Override
public void onPartitionsAssigned(String bindingName, Consumer<?, ?> consumer, Collection<TopicPartition> partitions, boolean initial) {

    Map<TopicPartition, Long> topicPartitionOffset = new HashMap<>();
    topicPartitionOffset.put(new TopicPartition("foo", 0), 1000L);
    topicPartitionOffset.put(new TopicPartition("foo", 1), 2000L);
    topicPartitionOffset.put(new TopicPartition("foo", 2), 2000L);
    topicPartitionOffset.put(new TopicPartition("foo", 3), 1000L);

    if (initial) {
        partitions.forEach(tp -> {
            if (topicPartitionOffset.containsKey(tp)) {
                final Long offset = topicPartitionOffset.get(tp);
                try {
                    consumer.seek(tp, offset);
                }
                catch (Exception e) {
                    // Handle exceptions carefully.
                }
            }
        });
    }
}

这只是一个基础实现。实际应用场景远比这复杂，你需要根据具体情况调整，但这无疑为你提供了一个基本框架。当消费者 seek 失败时，可能会抛出一些运行时异常，你需要决定在这些情况下如何处理。

[[what-if-we-start-a-second-consumer-with-the-same-group-id?]] === 如果我们启动第二个具有相同 group id 的消费者会怎样？

当我们添加第二个消费者时，会发生重新平衡，一些分区会被重新分配。假设新消费者获得了分区 2 和 3。当这个新的 Spring Cloud Stream 消费者调用 onPartitionsAssigned 方法时，它会看到这是该消费者上分区 2 和 3 的初始分配。因此，由于对 initial 参数的条件检查，它将执行 seek 操作。对于第一个消费者，它现在只拥有分区 0 和 1。然而，对于这个消费者来说，这只是一个重新平衡事件，并不被视为初始分配。因此，由于对 initial 参数的条件检查，它不会重新 seek 到给定的偏移量。

[[how-do-i-manually-acknowledge-using-kafka-binder?]] == 如何通过 Kafka binder 手动确认？

问题陈述

使用 Kafka binder 时，我想在消费者中手动确认消息。我该怎么做？

解决方案

默认情况下，Kafka绑定器委托给Spring for Apache Kafka项目中的默认提交设置。Spring Kafka中的默认ackMode是batch。有关此设置的更多详细信息，请参见此处。

在某些情况下，你可能希望禁用默认的提交行为，转而依赖手动提交。以下步骤可以帮助你实现这一点。

将属性 spring.cloud.stream.kafka.bindings.<binding-name>.consumer.ackMode 设置为 MANUAL 或 MANUAL_IMMEDIATE。这样设置后，消费者方法接收到的消息中将包含一个名为 kafka_acknowledgment（来自 KafkaHeaders.ACKNOWLEDGMENT）的头部信息。

例如，假设这是你的消费者方法。

@Bean
public Consumer<Message<String>> myConsumer() {
    return msg -> {
        Acknowledgment acknowledgment = message.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class);
        if (acknowledgment != null) {
         System.out.println("Acknowledgment provided");
         acknowledgment.acknowledge();
        }
    };
}

然后，你将属性 spring.cloud.stream.kafka.bindings.myConsumer-in-0.consumer.ackMode 设置为 MANUAL 或 MANUAL_IMMEDIATE。

[[how-do-i-override-the-default-binding-names-in-spring-cloud-stream?]] == 如何在 Spring Cloud Stream 中覆盖默认的绑定名称？

问题陈述

Spring Cloud Stream 会根据函数定义和签名创建默认绑定，但我该如何覆盖这些绑定以使用更具领域友好性的名称？

解决方案

假设以下是你的函数签名。

@Bean
public Function<String, String> uppercase(){
...
}

默认情况下，Spring Cloud Stream 将按如下方式创建绑定。

uppercase-in-0
uppercase-out-0

你可以通过以下属性将这些绑定覆盖为其他内容。

spring.cloud.stream.function.bindings.uppercase-in-0=my-transformer-in
spring.cloud.stream.function.bindings.uppercase-out-0=my-transformer-out

此后，所有绑定属性必须在新的名称 my-transformer-in 和 my-transformer-out 上进行设置。

这是另一个使用 Kafka Streams 和多个输入的例子。

@Bean
public BiFunction<KStream<String, Order>, KTable<String, Account>, KStream<String, EnrichedOrder>> processOrder() {
...
}

默认情况下，Spring Cloud Stream 会为这个函数创建三个不同的绑定名称。

processOrder-in-0
processOrder-in-1
processOrder-out-0

每次你想要在这些绑定上设置配置时，都必须使用这些绑定名称。你不喜欢这样，并且希望使用更具领域友好性和可读性的绑定名称，例如，类似这样的名称。

orders
accounts
enrichedOrders

你可以通过简单地设置这三个属性轻松实现这一点

spring.cloud.stream.function.bindings.processOrder-in-0=orders
spring.cloud.stream.function.bindings.processOrder-in-1=accounts
spring.cloud.stream.function.bindings.processOrder-out-0=enrichedOrders

一旦你这样做，它将覆盖默认的绑定名称，并且你想要设置的任何属性都必须基于这些新的绑定名称。

[[how-do-i-send-a-message-key-as-part-of-my-record?]] == 如何将消息键作为记录的一部分发送？

问题陈述

我需要随记录的有效负载一起发送一个键，Spring Cloud Stream 中有办法实现这一点吗？

解决方案

通常，您需要将类似映射的关联数据结构作为具有键和值的记录发送。Spring Cloud Stream 允许您以直接的方式实现这一点。以下是实现此功能的基本蓝图，但您可能需要根据具体用例进行调整。

这是一个示例生产者方法（也称为 Supplier）。

@Bean
public Supplier<Message<String>> supplier() {
    return () -> MessageBuilder.withPayload("foo").setHeader(KafkaHeaders.MESSAGE_KEY, "my-foo").build();
}

这是一个简单的函数，用于发送带有String负载和键的消息。请注意，我们使用KafkaHeaders.MESSAGE_KEY将键设置为消息头。

如果你想将键从默认的 kafka_messageKey 更改，那么在配置中，我们需要指定这个属性：

spring.cloud.stream.kafka.bindings.supplier-out-0.producer.messageKeyExpression=headers['my-special-key']

请注意，我们使用绑定名称 supplier-out-0，因为这是我们的函数名称，请相应地进行更新。

接着，我们在生成消息时使用这个新密钥。

[[how-do-i-use-native-serializer-and-deserializer-instead-of-message-conversion-done-by-spring-cloud-stream?]] == 如何在不使用 Spring Cloud Stream 的消息转换功能的情况下，使用原生的序列化器和反序列化器？

问题陈述

我不想使用 Spring Cloud Stream 中的消息转换器，而是想使用 Kafka 原生的序列化器和反序列化器。默认情况下，Spring Cloud Stream 会使用其内置的消息转换器来处理这种转换。我该如何绕过这一点，将责任委托给 Kafka？

解决方案

这真的很容易做到。

你只需提供以下属性即可启用原生序列化。

spring.cloud.stream.kafka.bindings.<binding-name>.producer.useNativeEncoding: true

然后，你还需要设置序列化器。有几种方法可以做到这一点。

spring.cloud.stream.kafka.bindings.<binding-name>.producer.configuration.key.serializer: org.apache.kafka.common.serialization.StringSerializer
spring.cloud.stream.kafka.bindings.<binding-name>.producer.configuration.value.serializer: org.apache.kafka.common.serialization.StringSerializer

或使用绑定器配置。

spring.cloud.stream.kafka.binder.configuration.key.serializer: org.apache.kafka.common.serialization.StringSerializer
spring.cloud.stream.kafka.binder.configuration.value.serializer: org.apache.kafka.common.serialization.StringSerializer

当使用binder方式时，它适用于所有绑定，而在绑定处设置则是针对每个绑定。

在反序列化端，您只需将反序列化器作为配置提供即可。

例如，

spring.cloud.stream.kafka.bindings.<binding-name>.consumer.configuration.key.deserializer: org.apache.kafka.common.serialization.StringDeserializer
spring.cloud.stream.kafka.bindings.<binding-name>.producer.configuration.value.deserializer: org.apache.kafka.common.serialization.StringDeserializer

你也可以在绑定器级别进行设置。

你可以设置一个可选属性来强制进行原生解码。

spring.cloud.stream.kafka.bindings.<binding-name>.consumer.useNativeDecoding: true

然而，对于Kafka绑定器而言，这是不必要的，因为当消息到达绑定器时，Kafka已经使用配置的反序列化器对它们进行了反序列化。

解释 Kafka Streams 绑定器中偏移量重置的工作原理

问题陈述

默认情况下，Kafka Streams绑定器总是从新消费者的最早偏移量开始。有时，从最新偏移量开始对应用程序是有益的或必需的。Kafka Streams绑定器允许您这样做。

解决方案

在查看解决方案之前，我们先来看以下场景。

@Bean
public BiConsumer<KStream<Object, Object>, KTable<Object, Object>> myBiConsumer{
    (s, t) -> s.join(t, ...)
    ...
}

我们有一个 BiConsumer bean，它需要两个输入绑定。在这种情况下，第一个绑定用于 KStream，第二个用于 KTable。首次运行此应用程序时，默认情况下，两个绑定都从 earliest 偏移量开始。如果由于某些需求，我想从 latest 偏移量开始，该怎么办？您可以通过启用以下属性来实现。

spring.cloud.stream.kafka.streams.bindings.myBiConsumer-in-0.consumer.startOffset: latest
spring.cloud.stream.kafka.streams.bindings.myBiConsumer-in-1.consumer.startOffset: latest

如果你希望只有一个绑定从 latest 偏移量开始消费，而另一个绑定使用默认的 earliest 偏移量，那么只需在配置中省略后者的绑定设置即可。

请注意，一旦存在已提交的偏移量，这些设置将不再生效，已提交的偏移量将优先。

跟踪 Kafka 中记录的成功发送（生产）

问题陈述

我有一个Kafka生产者应用程序，想要跟踪所有成功的发送记录。

解决方案

假设我们在应用程序中有以下供应商。

@Bean
	public Supplier<Message<String>> supplier() {
		return () -> MessageBuilder.withPayload("foo").setHeader(KafkaHeaders.MESSAGE_KEY, "my-foo").build();
	}

接着，我们需要定义一个名为 MessageChannel 的新 bean，用于捕获所有成功发送的信息。

@Bean
	public MessageChannel fooRecordChannel() {
		return new DirectChannel();
	}

接下来，在应用程序配置中定义此属性，以便为 recordMetadataChannel 提供 Bean 名称。

spring.cloud.stream.kafka.bindings.supplier-out-0.producer.recordMetadataChannel: fooRecordChannel

此时，成功发送的信息将被发送到 fooRecordChannel。

你可以编写一个 IntegrationFlow 如下所示来查看信息。

@Bean
public IntegrationFlow integrationFlow() {
    return f -> f.channel("fooRecordChannel")
                 .handle((payload, messageHeaders) -> payload);
}

在 handle 方法中，payload 是发送到 Kafka 的内容，而消息头包含一个名为 kafka_recordMetadata 的特殊键。其值是一个 RecordMetadata，其中包含有关主题分区、当前偏移量等信息。

在 Kafka 中添加自定义标头映射器

问题陈述

我有一个Kafka生产者应用程序设置了一些头部信息，但在消费者应用程序中这些头部信息缺失了。为什么会这样？

解决方案

在正常情况下，这应该没问题。

假设，你有以下生产者。

@Bean
public Supplier<Message<String>> supply() {
    return () -> MessageBuilder.withPayload("foo").setHeader("foo", "bar").build();
}

在消费者端，您应该仍然能看到标题 "foo"，并且以下内容不应给您带来任何问题。

@Bean
public Consumer<Message<String>> consume() {
    return s -> {
        final String foo = (String)s.getHeaders().get("foo");
        System.out.println(foo);
    };
}

如果你在应用程序中提供了自定义的头部映射器，那么这将无法正常工作。假设你在应用程序中有一个空的 KafkaHeaderMapper。

@Bean
public KafkaHeaderMapper kafkaBinderHeaderMapper() {
    return new KafkaHeaderMapper() {
        @Override
        public void fromHeaders(MessageHeaders headers, Headers target) {

        }

        @Override
        public void toHeaders(Headers source, Map<String, Object> target) {

        }
    };
}

如果这是你的实现，那么消费者将无法接收到 foo 标头。很可能，你在那些 KafkaHeaderMapper 方法中编写了一些逻辑。你需要以下步骤来填充 foo 标头。

@Bean
public KafkaHeaderMapper kafkaBinderHeaderMapper() {
    return new KafkaHeaderMapper() {
        @Override
        public void fromHeaders(MessageHeaders headers, Headers target) {
            final String foo = (String) headers.get("foo");
            target.add("foo", foo.getBytes());
        }

        @Override
        public void toHeaders(Headers source, Map<String, Object> target) {
            final Header foo = source.lastHeader("foo");
			target.put("foo", new String(foo.value()));
        }
    }

这将正确地将 foo 头部从生产者填充到消费者。

关于 id 标题的特殊说明

在 Spring Cloud Stream 中，id 头信息是一个特殊的头信息，但某些应用程序可能希望拥有特殊的自定义 id 头信息——例如 custom-id、ID 或 Id。第一个（custom-id）无需任何自定义头信息映射器即可从生产者传播到消费者。然而，如果你使用框架保留的 id 头信息的变体（如 ID、Id、iD 等）进行生产，那么你将会遇到框架内部的问题。有关此用例的更多背景信息，请参阅此 StackOverflow 主题。在这种情况下，你必须使用自定义的 KafkaHeaderMapper 来映射区分大小写的 id 头信息。例如，假设你有以下生产者。

@Bean
public Supplier<Message<String>> supply() {
    return () -> MessageBuilder.withPayload("foo").setHeader("Id", "my-id").build();
}

上方的 Id 标头在消费端将会消失，因为它与框架的 id 标头冲突。你可以提供一个自定义的 KafkaHeaderMapper 来解决这个问题。

@Bean
public KafkaHeaderMapper kafkaBinderHeaderMapper1() {
    return new KafkaHeaderMapper() {
        @Override
        public void fromHeaders(MessageHeaders headers, Headers target) {
            final String myId = (String) headers.get("Id");
			target.add("Id", myId.getBytes());
        }

        @Override
        public void toHeaders(Headers source, Map<String, Object> target) {
            final Header Id = source.lastHeader("Id");
			target.put("Id", new String(Id.value()));
        }
    };
}

通过这种方式，生产者和消费者两端都可以使用 id 和 Id 这两个头部信息。

在事务中向多个主题生产消息

问题陈述

如何向多个 Kafka 主题发送事务性消息？

更多背景信息，请参阅这个 StackOverflow 问题。

解决方案

在Kafka绑定器中使用事务支持来处理事务，然后提供一个AfterRollbackProcessor。为了向多个主题发送消息，请使用StreamBridge API。

以下是相关代码片段：

@Autowired
StreamBridge bridge;

@Bean
Consumer<String> input() {
    return str -> {
        System.out.println(str);
        this.bridge.send("left", str.toUpperCase());
        this.bridge.send("right", str.toLowerCase());
        if (str.equals("Fail")) {
            throw new RuntimeException("test");
        }
    };
}

@Bean
ListenerContainerCustomizer<AbstractMessageListenerContainer<?, ?>> customizer(BinderFactory binders) {
    return (container, dest, group) -> {
        ProducerFactory<byte[], byte[]> pf = ((KafkaMessageChannelBinder) binders.getBinder(null,
                MessageChannel.class)).getTransactionalProducerFactory();
        KafkaTemplate<byte[], byte[]> template = new KafkaTemplate<>(pf);
        DefaultAfterRollbackProcessor rollbackProcessor = rollbackProcessor(template);
        container.setAfterRollbackProcessor(rollbackProcessor);
    };
}

DefaultAfterRollbackProcessor rollbackProcessor(KafkaTemplate<byte[], byte[]> template) {
    return new DefaultAfterRollbackProcessor<>(
            new DeadLetterPublishingRecoverer(template), new FixedBackOff(2000L, 2L), template, true);
}

必需配置

spring.cloud.stream.kafka.binder.transaction.transaction-id-prefix: tx-
spring.cloud.stream.kafka.binder.required-acks=all
spring.cloud.stream.bindings.input-in-0.group=foo
spring.cloud.stream.bindings.input-in-0.destination=input
spring.cloud.stream.bindings.left.destination=left
spring.cloud.stream.bindings.right.destination=right

spring.cloud.stream.kafka.bindings.input-in-0.consumer.maxAttempts=1

为了进行测试，你可以使用以下方法：

@Bean
public ApplicationRunner runner(KafkaTemplate<byte[], byte[]> template) {
    return args -> {
        System.in.read();
        template.send("input", "Fail".getBytes());
        template.send("input", "Good".getBytes());
    };
}

一些重要说明：

请确保在应用程序配置中没有设置任何DLQ（死信队列），因为我们手动配置了DLT（默认情况下，基于初始消费者函数，消息将被发布到名为input.DLT的主题）。同时，将消费者绑定上的maxAttempts重置为1，以避免绑定器进行重试。在上面的示例中，消息最多会被尝试处理3次（初始尝试 + FixedBackoff中的2次重试）。

有关如何测试此代码的更多详情，请参阅StackOverflow讨论帖。如果您使用 Spring Cloud Stream 通过添加更多消费者函数来测试，请确保将消费者绑定上的 isolation-level 设置为 read-committed。

这个StackOverflow 讨论也与本次讨论相关。

运行多个可轮询消费者时需要避免的陷阱

问题陈述

如何运行多个可轮询消费者的实例，并为每个实例生成唯一的 client.id？

解决方案

假设我有以下定义：

spring.cloud.stream.pollable-source: foo
spring.cloud.stream.bindings.foo-in-0.group: my-group

在运行应用程序时，Kafka消费者会生成一个client.id（类似于consumer-my-group-1）。对于每个正在运行的应用程序实例，这个client.id将是相同的，从而导致意外问题。

为了解决这个问题，你可以在应用程序的每个实例上添加以下属性：

spring.cloud.stream.kafka.bindings.foo-in-0.consumer.configuration.client.id=${client.id}

更多详情请参阅此 GitHub issue。

使用 Kafka 的简单 DLQ​

问题陈述​

解决方案​

具有高级重试选项的 DLQ​

问题陈述​

解决方案​

使用 DLQ 处理反序列化错误​

问题陈述​

解决方案​

Kafka绑定器中的基本偏移量管理​

问题陈述​

解决方案​

在 Kafka 中定位到任意偏移量​

问题陈述​

解决方案​

问题陈述​

解决方案​

问题陈述​

解决方案​

问题陈述​

解决方案​

问题陈述​

解决方案​

解释 Kafka Streams 绑定器中偏移量重置的工作原理​

问题陈述​

解决方案​

跟踪 Kafka 中记录的成功发送（生产）​

问题陈述​

解决方案​

在 Kafka 中添加自定义标头映射器​

问题陈述​

解决方案​

关于 id 标题的特殊说明​

在事务中向多个主题生产消息​

问题陈述​

解决方案​

必需配置​

运行多个可轮询消费者时需要避免的陷阱​

问题陈述​

解决方案​

使用 Kafka 的简单 DLQ

问题陈述

解决方案

具有高级重试选项的 DLQ

问题陈述

解决方案

使用 DLQ 处理反序列化错误

问题陈述

解决方案

Kafka绑定器中的基本偏移量管理

问题陈述

解决方案

在 Kafka 中定位到任意偏移量

问题陈述

解决方案

问题陈述

解决方案

问题陈述

解决方案

问题陈述

解决方案

问题陈述

解决方案

解释 Kafka Streams 绑定器中偏移量重置的工作原理

问题陈述

解决方案

跟踪 Kafka 中记录的成功发送（生产）

问题陈述

解决方案

在 Kafka 中添加自定义标头映射器

问题陈述

解决方案

关于 id 标题的特殊说明

在事务中向多个主题生产消息

问题陈述

解决方案

必需配置

运行多个可轮询消费者时需要避免的陷阱

问题陈述

解决方案