配置AggregatingReplyingKafkatTemplate以获取来自多个主题和多个使用者实例的响应

Question

这是我发布的另一个问题的延续。以下是之前问题的链接：来自多个主题的消息聚合

我的要求是确保node1应该接收响应R1M1

实施的事情：

>

设置Kafka领导人。来自两个节点的生产者记录中的相关ID。

KafkCamessageListenerContainerBean使用容器属性中的两个响应主题创建。ContainerProperties ContainerProperties=新的ContainerProperty（output-message-topic1，output-message-topic2）；

聚合回复KafkatTemplate。setSharedReplyTopic（true）；

consumerFactory配置为单个groupId。

 consumerConfigProps.put(ConsumerConfig.GROUP_ID_CONFIG, "xxxx.yyyyy.zzzz");

注:

>

由于infra的限制，无法为每个使用者（节点）创建单独的主题。

我需要关于需要实现的更改的帮助，以获得请求的相关聚合响应。

Answer 1

看到留档。

您必须将每个实例配置为使用离散的回复主题分区，并且服务器必须将回复路由到请求的分区；或者您必须使用不同的组。每个实例的id，以便向这两个实例发送回复（并被未发出请求的实例丢弃）。

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

如果您有多个客户端实例，并且您没有如上一段所述配置它们，则每个实例都需要一个专用的回复主题。另一种选择是设置KafkaHeaders。REPLY_PARTITION并为每个实例使用专用分区。Header包含一个四个字节的int（big-endian）。服务器必须使用此标头将回复路由到正确的分区（@KafkaListener执行此操作）。然而，在这种情况下，回复容器不能使用Kafka的组管理功能，必须配置为在固定分区上侦听（通过在其构造函数中使用ContainerProperties中的TopicPARtionOffset）。