根据这个主题:
kafka-spring集成:kafka消费者不提供头文件-这是对kafka的无头文件支持
但文件上说:
spring.cloud.stream.kafka.binder.headers文件
将由活页夹传送的自定义标头的列表。
默认值:空。
我无法使用SpringCloudStreamBinder kafka:1.2.0.release
发送日志:
MESSAGE (e23885fd-ffd9-42dc-ebe3-5a78467fee1f) SENT :
GenericMessage [payload=...,
headers={
content-type=application/json,
correlationId=51dd90b1-76e6-4b8d-b667-da25f214f383,
id=e23885fd-ffd9-42dc-ebe3-5a78467fee1f,
contentType=application/json,
timestamp=1497535771673
}]
接收日志:
MESSAGE (448175f5-2b21-9a44-26b9-85f093b33f6b) RECEIVED BY HANDLER 1:
GenericMessage [payload=...,
headers={
kafka_offset=36,
id=448175f5-2b21-9a44-26b9-85f093b33f6b,
kafka_receivedPartitionId=0,
contentType=application/json;charset=UTF-8,
kafka_receivedTopic=new_patient, timestamp=1497535771715
}]
MESSAGE (448175f5-2b21-9a44-26b9-85f093b33f6b) RECEIVED BY HANDLER 2 :
GenericMessage [payload=...,
headers={
kafka_offset=36,
id=448175f5-2b21-9a44-26b9-85f093b33f6b,
kafka_receivedPartitionId=0,
contentType=application/json;charset=UTF-8,
kafka_receivedTopic=new_patient, timestamp=1497535771715
}]
我希望看到相同的消息id并在接收端获得correlationid。
应用程序属性:
spring.cloud.stream.kafka.binder.headers=correlationId
spring.cloud.stream.bindings.newTest.destination=new_test
spring.cloud.stream.bindings.newTestCreated.destination=new_test
spring.cloud.stream.default.consumer.headerMode=embeddedHeaders
spring.cloud.stream.default.producer.headerMode=embeddedHeaders
正在发送消息:
@Publisher(channel = "testChannel")
public Object newTest(Object param) {
...
return myObject;
}
1条答案
按热度按时间neekobn81#
是的,它有:http://docs.spring.io/spring-cloud-stream/docs/chelsea.sr2/reference/htmlsingle/index.html#_consumer_properties
头模
当设置为raw时,禁用对输入的头解析。仅对本机不支持消息头并且需要嵌入消息头的消息传递中间件有效。当入站数据来自外部spring云流应用程序时非常有用。
默认值:embeddedheaders
但那已经是Spring的云流故事,而不是Spring的Kafka本身。