我正在使用 Kafka Spring Integration 使用 kafka 发布和消费消息。我看到有效负载已正确从生产者传递到消费者,但标头信息在某处被覆盖。
@ServiceActivator(inputChannel = "fromKafka")
public void processMessage(Message<?> message) throws InterruptedException,
ExecutionException {
try {
System.out.println("Headers :" + message.getHeaders().toString());
}
} catch (Exception e) {
e.printStackTrace();
}
}
我得到以下标题:
Headers :{timestamp=1440013920609, id=f8c645f7-677b-ec32-dad0-a7b79082ef81}
我在生产者端构造消息,如下所示:
Message<FeelDBMessage> message = MessageBuilder
.withPayload(samplePayloadObj)
.setHeader(KafkaHeaders.MESSAGE_KEY, "key")
.setHeader(KafkaHeaders.TOPIC, "sampleTopic").build();
// publish the message
publisher.publishMessage(message);
下面是生产者的标头信息:
headers={timestamp=1440013914085, id=c4159c1c-2c67-634b-ef8d-3fb026b1172e, kafka_messageKey=key, kafka_topic=sampleTopic}
知道为什么标头被不同的值覆盖吗?
只是因为默认情况下框架使用不可变的 GenericMessage
.
对现有消息的任何操作(例如MessageBuilder.withPayload
)将产生一个新的GenericMessage
实例。
从另一方面来看,卡夫卡不支持任何headers
像 JMS 或 AMQP 这样的抽象。这就是为什么KafkaProducerMessageHandler
只需在向 Kafka 发布消息时执行此操作:
this.kafkaProducerContext.send(topic, partitionId, messageKey, message.getPayload());
如您所见,它没有发送headers
根本不。所以,另一方(消费者)只处理message
从主题作为payload
和一些系统选项headers
like topic
, partition
, messageKey
.
简而言之:我们不通过 Kafka 传输标头,因为它不支持它们。
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)