spring集成:Kafka消费者的头不来了

k3bvogb1  于 2021-06-08  发布在  Kafka
关注(0)|答案(1)|浏览(343)

我正在使用kafka-spring集成来使用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);

下面是producer的标题信息:

headers={timestamp=1440013914085, id=c4159c1c-2c67-634b-ef8d-3fb026b1172e, kafka_messageKey=key, kafka_topic=sampleTopic}

知道为什么头会被另一个值覆盖吗?

j0pj023g

j0pj023g1#

只是因为默认情况下框架使用的是不可变的 GenericMessage .
对现有消息的任何操作(例如。 MessageBuilder.withPayload )将产生一个新的 GenericMessage 示例。
另一方面,Kafka不支持任何 headers 像jms或amqp这样的抽象。这就是为什么 KafkaProducerMessageHandler 当它向Kafka发布消息时,请执行以下操作:

this.kafkaProducerContext.send(topic, partitionId, messageKey, message.getPayload());

如你所见,它没有发送 headers 完全。所以,另一方(消费者)只处理 message 从主题作为 payload 以及一些系统选项 headers 就像 topic , partition , messageKey .
换句话说:我们不通过Kafka传输头文件,因为它不支持头文件。

相关问题