Kafka Consumer获得关键值对

Question

我目前正在与Kafka和Flink合作,我在我的本地PC上运行了kafka,并创建了一个正在消费的主题.

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

但它只是检索消息,

有没有办法获得有关该消息的更多细节？让我们说时间？键？我检查了kafka文档,但我没有找到关于这个主题的东西

Answer 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

Answer 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