在Kafka消息中添加自定义标头

Question

我通过使用kafka生成器将文件转换为字节数组来发送文件作为消息.

我还需要为消息添加一些标题,例如文件名,时间戳等,因此在消费者端我可以根据文件名和其他标题处理消息.

我目前正在做的是创建一个对象并将原始消息和标头包装在其中,并将字节数组中的对象作为消息发送.

我想知道是否有一种方法可以在发布消息时添加自定义标头？

Answer 1

Kafka v0.11.0.0增加了对自定义标头的支持.

您可以在创建ProducerRecord时添加它们,如下所示:

new ProducerRecord(key,value,headers,...),其中headers的类型为Iterable <Header>

有关更多详情,请参阅

https://issues.apache.org/jira/browse/KAFKA-4208

https://cwiki.apache.org/confluence/display/KAFKA/KIP-82+-+Add+Record+Headers

Answer 2

记录级标头是从 Kafka 0.11.0 引入的。我们可以在每个记录中发送一个标题列表。

List<Header> headers = Arrays.asList(new RecordHeader("header_key", "header_value".getBytes()));
ProducerRecord<String, String> record = new ProducerRecord<>("topic", null, "key", "value", headers);

Answer 3

Kafka 对消息内容是不可知的，并且不提供任何特殊的方法来丰富它，因此这是您需要自己做的事情。处理这些事情的常见方法是使用结构化格式，例如 json、avro 或类似的格式，您可以自由定义必要的字段，并且可以轻松地将元数据添加到消息中并将其发送到 Kafka 代理。

从 Kafka 0.11 开始，这个答案已经过时，请参阅其他答案。