在ApacheKafkaJava中实现请求-响应

Question

请找到我们需要实现的用例。

首先，我们需要调用Kafka生产者将消息作为rest服务，他们将在另一个主题中处理并返回响应。

对于我们来说，这是一个请求-回复主题，我们需要回复相同的请求-响应，使用replykafka模板工作正常，但我们可以在标题中设置相关id。

作为主题消息元数据，存在发送属性，是否有任何方法将关联id映射到请求主题消息和回复主题消息。

给你好好解释一下。

一个微服务期望负载如下所示，负载中包含correlationId。

{
  "operationDate": "2020-09-16T11:58:25",
  "correlationId": "-5544538377183901824042719876882142227",
  "birthDate": "2013-12-12",
  "firstNameEn": "boby",
  "firstNameAr": "الشيخ",
}

微服务将处理有效负载，并在另一个主题中给出响应。

{
  "correlationId": -5544538377183901824042719876882142227,
  "consumerId": null,
  "userid": 123456,
  "statusCode": "SUCCESS",
  "errors": null
}

现在，我们需要实现使用Spring ReplyingKafkaTem板。

作为回复，KafkatTemplate将仅与标头中的correlationId一起使用

Answer 1

谢谢你的提示。

我已经做了与Kafka头相关的有效载荷覆盖。

@Override
    protected ListenableFuture<SendResult> doSend(ProducerRecord producerRecord) {
        if(producerRecord.value()!=null){
           // i have appeneded the header correlationId in th payload
        }
        return super.doSend(producerRecord);
    }

在Replay onMessage中，我已将响应负载correlationId填充到头中。

@Override
    public void onMessage(List<ConsumerRecord<K, R>> data) {
        data.forEach(
            krConsumerRecord -> //update each record header
        );
        super.onMessage(data);
    }

通过这种方式，成功地将请求-响应语义与请求和响应负载中的correlationId集成在一起。

Answer 2

假设您的意思是希望在相关id中包含主题，请参阅

java prettyprint-override">/**
 * Set a function to be called to establish a unique correlation key for each request
 * record.
 * @param correlationStrategy the function.
 * @since 2.3
 */
public void setCorrelationIdStrategy(Function<ProducerRecord<K, V>, CorrelationKey> correlationStrategy) {

您可以创建自己的相关ID，基于producerRecords（其中包含主题（））。

你只需要确保它是独一无二的。如果您手动设置KafkaHeaders。REPLY_TOPIC，它将对策略可见。

编辑

使用有效负载中的相关id，使用setCorrelationIdStrategy从有效负载中提取相关id，并添加RecordInterceptor在应答端执行相同操作。