发送消息

Doubao 中英对照 Sending Messages

本节介绍如何发送消息。

使用 KafkaTemplate

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

概述

KafkaTemplate 包装了一个生产者，并提供向 Kafka 主题发送数据的便捷方法。以下清单展示了 KafkaTemplate 中的相关方法：

CompletableFuture<SendResult<K, V>> sendDefault(V data);

CompletableFuture<SendResult<K, V>> sendDefault(K key, V data);

CompletableFuture<SendResult<K, V>> sendDefault(Integer partition, K key, V data);

CompletableFuture<SendResult<K, V>> sendDefault(Integer partition, Long timestamp, K key, V data);

CompletableFuture<SendResult<K, V>> send(String topic, V data);

CompletableFuture<SendResult<K, V>> send(String topic, K key, V data);

CompletableFuture<SendResult<K, V>> send(String topic, Integer partition, K key, V data);

CompletableFuture<SendResult<K, V>> send(String topic, Integer partition, Long timestamp, K key, V data);

CompletableFuture<SendResult<K, V>> send(ProducerRecord<K, V> record);

CompletableFuture<SendResult<K, V>> send(Message<?> message);

Map<MetricName, ? extends Metric> metrics();

List<PartitionInfo> partitionsFor(String topic);

<T> T execute(ProducerCallback<K, V, T> callback);

<T> T executeInTransaction(OperationsCallback<K, V, T> callback);

// Flush the producer.
void flush();

interface ProducerCallback<K, V, T> {

    T doInKafka(Producer<K, V> producer);

}

interface OperationsCallback<K, V, T> {

    T doInOperations(KafkaOperations<K, V> operations);

}

更多细节请参阅 Javadoc。

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

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

metrics 和 partitionsFor 方法委托给底层 Producer 上的相同方法。execute 方法提供对底层 Producer 的直接访问。

备注

从 Spring 组件发送消息时，如果依赖通过 NewTopic bean 自动创建主题，请避免使用 @PostConstruct 方法。@PostConstruct 在应用程序上下文完全就绪之前运行，这可能在干净的代理上导致 UnknownTopicOrPartitionException。

相反，请考虑以下替代方案：

• 使用 ApplicationListener<ContextRefreshedEvent> 以确保在发送之前上下文已完全刷新。
• 实现 SmartLifecycle，以便在 KafkaAdmin bean 完成初始化之后启动。
• 预先在外部创建主题。

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

@Bean
public ProducerFactory producerFactory() {
    return new DefaultKafkaProducerFactory<>(producerConfigs());
}

@Bean
public Map producerConfigs() {
    Map props = new HashMap<>();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    // See https://kafka.apache.org/42/documentation/#producerconfigs for more properties
    return props;
}

@Bean
public KafkaTemplate kafkaTemplate() {
    return new KafkaTemplate(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 Boot 自动配置的那个）可以用不同的收窄泛型类型引用。

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

然后，要使用该模板，你可以调用其方法之一。

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

• KafkaHeaders.TOPIC
• KafkaHeaders.PARTITION
• KafkaHeaders.KEY
• KafkaHeaders.TIMESTAMP

消息载荷就是数据。

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

public interface ProducerListener<K, V> {

    default void onSuccess(ProducerRecord<K, V> producerRecord, RecordMetadata recordMetadata) {
	}

    default void onError(ProducerRecord<K, V> producerRecord, RecordMetadata recordMetadata, Exception exception) {
	}

}

默认情况下，模板配置有一个 LoggingProducerListener，它记录错误，并在发送成功时不做任何事。

为方便起见，提供了默认方法实现，以防你只想实现其中一个方法。

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

CompletableFuture<SendResult<Integer, String>> future = template.send("myTopic", "something");
future.whenComplete((result, ex) -> {
    ...
});

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

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

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

示例

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

示例 1：非阻塞（异步）

public void sendToKafka(final MyOutputData data) {
    final ProducerRecord<String, String> record = createRecord(data);

    CompletableFuture<SendResult<String, String>> future = template.send(record);
    future.whenComplete((result, ex) -> {
        if (ex == null) {
            handleSuccess(data);
        }
        else {
            handleFailure(data, record, ex);
        }
    });
}

示例 2：阻塞（同步）

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 的 cause 是带有 producerRecord 属性的 KafkaProducerException。

使用 RoutingKafkaTemplate

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

important

路由模板 不 支持事务、execute、flush 或 metrics 操作，因为这些操作的主题未知。

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

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

@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("bytesPF", DefaultKafkaProducerFactory.class, () -> 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 见 注解属性。

有关实现类似结果、但还具有向同一主题发送不同类型这一额外能力的另一种技术，请参阅 委托序列化器和反序列化器。

使用 DefaultKafkaProducerFactory

如 使用 KafkaTemplate 所见，ProducerFactory 用于创建生产者。

不使用 事务 时，默认情况下，DefaultKafkaProducerFactory 创建一个由所有客户端使用的单例生产者，正如 KafkaProducer JavaDocs 中推荐的那样。但是，如果你在模板上调用 flush()，这可能会给使用同一生产者的其他线程带来延迟。从 2.3 版本开始，DefaultKafkaProducerFactory 新增了一个属性 producerPerThread。设置为 true 时，工厂将为每个线程创建（并缓存）一个单独的生产者，以避免此问题。

important

当 producerPerThread 为 true 时，用户代码 必须 在生产者不再需要时，在工厂上调用 closeThreadBoundProducer()。这将物理关闭该生产者并将其从 ThreadLocal 中移除。调用 reset() 或 destroy() 不会清理这些生产者。

另请参阅 KafkaTemplate 事务性与非事务性发布。

创建 DefaultKafkaProducerFactory 时，可以通过调用只接收属性 Map 的构造函数从配置中获取键和/或值 Serializer 类（见 使用 KafkaTemplate 中的示例），或者将 Serializer 实例传递给 DefaultKafkaProducerFactory 构造函数（在这种情况下，所有 Producer 共享相同实例）。或者，你可以提供 Supplier<Serializer>（自 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);

（另请参阅 使用 Message<?> 的请求/回复。）

结果是一个 CompletableFuture，它被异步填充结果（或超时情况下的异常）。该结果还有一个 sendFuture 属性，即调用 KafkaTemplate.send() 的结果。你可以使用此 future 来确定发送操作的结果。

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

从 2.8.8 版本开始，模板新增了一个方法 waitForAssignment。如果回复容器配置为 auto.offset.reset=latest，以避免在容器初始化之前发送请求和回复，这会很有用。

important

使用手动分区分配（无消费者组管理）时，等待的时长必须大于容器的 pollTimeout 属性，因为通知要到第一次 poll 完成之后才会发送。

以下 Spring Boot 应用展示了如何使用该特性：

@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 -> {
            if (!template.waitForAssignment(Duration.ofSeconds(10))) {
                throw new IllegalStateException("Reply container did not initialize");
            }
            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。如此配置后，RequestReplyFuture 将以异常完成，你可以捕获 ExecutionException，其 cause 属性中带有 DeserializationException。

从 2.6.7 版本开始，除了检测 DeserializationException 之外，如果提供了 replyErrorChecker 函数，模板还会调用它。如果它返回一个异常，该 future 将以异常完成。

以下是一个示例：

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，它会把意外回复的日志级别从默认的 ERROR 降为 DEBUG。

以下是配置回复容器以使用同一共享回复主题的示例：

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

important

如果你有多个客户端实例，且未按上一段所述配置它们，每个实例需要一个专用回复主题。另一种方法是设置 KafkaHeaders.REPLY_PARTITION 并为每个实例使用专用分区。该 Header 包含一个四字节 int（大端序）。服务端必须使用此头将回复路由到正确分区（@KafkaListener 会这样做）。但在这种情况下，回复容器不得使用 Kafka 的消费者组管理功能，而必须配置为监听固定分区（在其 ContainerProperties 构造函数中使用 TopicPartitionOffset）。

备注

JsonKafkaHeaderMapper 要求 Jackson 在 classpath 上（对于 @KafkaListener）。如果不可用，消息转换器就没有头映射器，因此你必须用 SimpleKafkaHeaderMapper 配置一个 MessagingMessageConverter，如前文所示。

默认使用 3 个头：

• KafkaHeaders.CORRELATION_ID——用于将回复关联到请求
• KafkaHeaders.REPLY_TOPIC——用于告知服务端回复到何处
• KafkaHeaders.REPLY_PARTITION——（可选）用于告知服务端回复到哪个分区

这些头名称由 @KafkaListener 基础设施用于路由回复。

从 2.3 版本开始，你可以自定义头名称——模板有 3 个属性 correlationHeaderName、replyTopicHeaderName 和 replyPartitionHeaderName。如果你的服务端不是 Spring 应用（或不使用 @KafkaListener），这会很有用。

备注

反过来，如果发起请求的应用不是 Spring 应用，且把关联信息放在不同的头中，那么从 3.0 版本开始，你可以在监听容器工厂上配置自定义的 correlationHeaderName，该头将被回显。此前，监听器必须回显自定义关联头。

使用 Message<?> 的请求/回复

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 应用时。以下是后一种情况的示例：

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 ByteArrayJacksonJsonMessageConverter());
    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.setGroupId("request.replies")
    val template = ReplyingKafkaTemplate<String, String, String>(pf, replyContainer)
    template.messageConverter = ByteArrayJacksonJsonMessageConverter()
    template.setDefaultTopic("requests")
    return template
}

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 { thing1 -> log.info(thing1.toString()) }

回复类型 Message<?>

当 @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.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.KEY, 42)
            .build();
}

回复中的原始记录键

从 3.3 版本开始，来自入站请求的 Kafka 记录键（如果存在）将保留在回复记录中。这仅适用于单记录请求/回复场景。当监听器是批量的或返回类型是集合时，由应用通过将回复记录包装在 Message 类型中来指定使用哪些键。

聚合多个回复

使用 ReplyingKafkaTemplate 中的模板严格用于单请求/回复场景。对于单条消息的多个接收方都返回回复的情况，你可以使用 AggregatingReplyingKafkaTemplate。它是 散射-聚集企业集成模式 客户端侧的实现。

与 ReplyingKafkaTemplate 类似，AggregatingReplyingKafkaTemplate 构造函数接收一个生产者工厂和一个监听容器来接收回复；它有第三个参数 BiPredicate<List<ConsumerRecord<K, R>>, Boolean> releaseStrategy，每次收到回复时都会咨询它；当谓词返回 true 时，该 ConsumerRecord 集合被用于完成 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 集合。"外层" ConsumerRecord 不是一条 "真实" 记录，它由模板合成，作为为该请求收到的实际回复记录的持有者。当正常释放发生（释放策略返回 true）时，主题被设为 aggregatedResults；如果 returnPartialOnTimeout 为 true 且发生超时（且至少收到一条回复记录），主题被设为 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 包含从中收到回复的实际主题。

important

用于回复的监听容器 必须 配置为 AckMode.MANUAL 或 AckMode.MANUAL_IMMEDIATE；消费者属性 enable.auto.commit 必须为 false（自 2.3 版本起的默认值）。为避免任何丢失消息的可能，模板仅在没有未完成请求时（即当最后一个未完成请求被释放策略释放时）才提交偏移量。再均衡之后，可能出现重复回复投递；这些会被任何在途请求忽略；当对已释放的回复收到重复回复时，你可能会看到错误日志消息。

备注

如果你在此聚合模板中使用 ErrorHandlingDeserializer，框架不会自动检测 DeserializationException。相反，记录（值为 null）会原样返回，反序列化异常放在头中。建议应用调用工具方法 ReplyingKafkaTemplate.checkDeserialization() 以判断是否发生了反序列化异常。更多信息请参阅其 JavaDocs。此聚合模板也不会调用 replyErrorChecker；你应对回复的每个元素执行检查。