Kafka Spring集成:标头不适用于kafka消费者

问题描述

我正在使用Kafka Spring Integration通过kafka发布和使用消息.我看到有效负载已正确地从生产者传递到消费者，但是标头信息在某处被覆盖.

I am using Kafka Spring Integration for publishing and consuming messages using kafka. I see Payload is properly passed from producer to consumer, but the header information is getting overridden somewhere.

@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}

我正在像这样在生产者端构造消息:

I am constructing the message at producer end like this:

Message<FeelDBMessage> message = MessageBuilder
                .withPayload(samplePayloadObj)
                .setHeader(KafkaHeaders.MESSAGE_KEY, "key")
                .setHeader(KafkaHeaders.TOPIC, "sampleTopic").build();

        // publish the message
        publisher.publishMessage(message);

及以下是生产者的标题信息:

and below is the header info at producer:

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

您知道为什么标头会被其他值覆盖吗?

Any idea why the Headers are overridden by a different value?

推荐答案

只是因为默认情况下Framework使用不可变 GenericMessage.

Just because by default Framework uses the immutable GenericMessage.

对现有消息(例如MessageBuilder.withPayload)进行任何操作都会产生一个新的GenericMessage实例.

Any manipulation to the existing message (e.g. MessageBuilder.withPayload) will produce a new GenericMessage instance.

另一方面，Kafka不支持JMS或AMQP之类的任何headers抽象.这就是KafkaProducerMessageHandler在向Kafka发布消息时执行此操作的原因:

From other side Kafka doesn't support any headers abstraction like JMS or AMQP. That's why KafkaProducerMessageHandler just do this when it publishes a message to Kafka:

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

如您所见，它根本不发送headers.因此，另一面(消费者)仅将主题中的message作为payload处理，并将某些系统选项作为headers处理，例如topic，partition，messageKey.

As you see it doesn't send headers at all. So, other side (consumer) just deals with only message from the topic as a payload and some system options as headers like topic, partition, messageKey.

两个词:我们不通过Kafka传输标头，因为它不支持标头.

In two words: we don't transfer headers over Kafka because it doesn't support them.

这篇关于Kafka Spring集成:标头不适用于kafka消费者的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！