提示、技巧和食谱

使用 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，应用程序必须提供一个组名称。匿名消费者无法使用 DLQ 功能。我们还需要通过将 Kafka 消费者绑定上的 enableDLQ 属性设置为 true 来启用 DLQ。最后，我们可以选择通过在 Kafka 消费者绑定上提供 dlqName 来提供 DLT 名称，否则在此情况下默认为 error.input-topic.my-group。

请注意，在上面提供的示例消费者中，有效负载的类型为 byte[]。默认情况下，Kafka 绑定器中的 DLQ 生产者期望 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 绑定器中内置的默认重试选项。

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

spring.cloud.stream.bindings.processData-in-0.consumer.maxAtttempts
spring.cloud.stream.bindings.processData-in-0.consumer.backOffInitialInterval
spring.cloud.stream.bindings.processData-in-0.consumer.backOffMultipler
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，则可以使用以下属性指定您的绑定所需的 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 项目提供了一些很好的方法来帮助绑定程序处理这种情况。让我们来探索一下。

假设这是我们的函数

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

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

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

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

为了处理这种情况，您可以添加一个配置了DefaultErrorHandler的ListenerContainerCustomizer @Bean。此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);
	}

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

第二个 bean 是DefaultErrorHandler，它配置为发布到DLT。有关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 应用程序默认情况下将从主题分区中可用的latest偏移量开始。另一方面，如果您显式指定了 group.id，那么默认情况下，Spring Cloud Stream 应用程序将从主题分区中可用的earliest偏移量开始。

在上述两种情况下（具有显式组的消费者和匿名组），可以通过使用属性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，它将注入到所有消费者绑定中。这是一个带有几个默认方法的接口，但这里是我们感兴趣的方法

/**
	 * 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

0           1000
1           2000
2           2000
3           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?]] === 如果我们使用相同的组 ID 启动第二个消费者会怎样？

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

[[how-do-i-manually-acknowledge-using-kafka-binder?]] == 如何使用 Kafka 绑定程序手动确认？

问题陈述

使用 Kafka 绑定程序，我想在消费者中手动确认消息。我该怎么做？

解决方案

默认情况下，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

一旦你这么做了，它就会覆盖默认的绑定名称，并且你想在上面设置的任何属性都必须在这些新的绑定名称上。

[[如何将消息键作为记录的一部分发送？]] == 如何将消息键作为记录的一部分发送？

问题陈述

我需要将一个键与记录的有效负载一起发送，在 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，因为这是我们的函数名称，请相应更新。

然后，我们在生成消息时使用此新键。

[[如何使用原生序列化器和反序列化器而不是 Spring Cloud Stream 完成的消息转换？]] == 如何使用原生序列化器和反序列化器而不是 Spring Cloud Stream 完成的消息转换？

问题陈述

我想在 Kafka 中使用原生序列化器和反序列化器，而不是在 Spring Cloud Stream 中使用消息转换器。默认情况下，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

使用绑定器方式时，它将应用于所有绑定，而绑定设置则针对每个绑定。

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

例如，

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 方法中，有效负载是发送到 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 问题。