Kafka Consumer 获取键值对

Posted 2023-03-10

【中文标题】Kafka Consumer 获取键值对

【英文标题】：Kafka Consumer get key value pair

【发布时间】：2017-05-14 23:45:50

【问题描述】：

我目前正在使用 Kafka 和 Flink，我在本地 PC 上运行了 kafka，并且我创建了一个正在使用的主题。

桌面\kafka\bin\windows>kafka-console-consumer.bat --zookeeper localhost:2181 -topic test

但它只是检索消息，

有没有办法获得有关消息的更多详细信息？让我们说时间？钥匙？我检查了 kafka 文档，但没有找到有关此主题的内容

【参考方案1】：

使用开箱即用的控制台消费者（我使用的是 Kafka 0.9.0.1），您只能使用不同格式打印消息的键和值。要打印密钥，请设置属性print.key=true。

还有另一个属性key.separator，默认情况下是“\t”（一个选项卡），您也可以将其更改为您想要的任何内容。

要设置这些属性，您可以创建一个配置文件并使用--consumer.config <config file> 或使用--property key=value 传递属性。

您也可以实现自己的格式化程序并将其与--formatter 选项一起使用，但您仍然只有键和值，因为这是 MessageFormatter 特征提供的（参见下面的 writeTo）。

trait MessageFormatter 
    def writeTo(key: Array[Byte], value: Array[Byte], output: PrintStream)

    def init(props: Properties) 

    def close() 

例如：

./bin/kafka-console-consumer.sh --new-consumer --bootstrap-server kafka-1:9092 --topic topic1 --property print.key=true --property key.separator="-" --from-beginning
key-p1
key-p2
key-p3
null-4
null-8
null-0

【讨论】：

我不想提出一个新问题，话虽如此，我认为 Kafka 应该提供一种方法来为每条消息或至少为每个主题指定生产者的密钥。你知道我可以用什么吗？

$ ./bin/kafka-console-producer.sh --broker-list kafka-1:9092 --topic topic1 --property parse.key=true --property key.separator=" -" 关键-价值

为我解决了问题..

【参考方案2】：

使用以下命令：

kafka-console-consumer --bootstrap-server localhost:9092 --topic topic_name \ 
      --from-beginning --formatter kafka.tools.DefaultMessageFormatter \
      --property print.key=true --property print.value=true \
      --property key.deserialzer=org.apache.kafka.common.serialization.StringDeserializer \
      --property value.deserializer=org.apache.kafka.common.serialization.LongDeserializer

【讨论】：

Windows 用户注意 => 对于 cmd 将 \ 替换为 ^