我想要生产者在Kafka主题中插入消息的时间戳。
在Kafka消费端,我想提取时间戳。
class Producer {
def main(args: Array[String]): Unit = {
writeToKafka("quick-start")
}
def writeToKafka(topic: String): Unit = {
val props = new Properties()
props.put("bootstrap.servers", "localhost:9094")
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
val producer = new KafkaProducer[String, String](props)
val record = new ProducerRecord[String, String](topic, "key", "value")
producer.send(record)
producer.close()
}
}
class Consumer {
def main(args: Array[String]): Unit = {
consumeFromKafka("quick-start")
}
def consumeFromKafka(topic: String) = {
val props = new Properties()
props.put("bootstrap.servers", "localhost:9094")
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
props.put("auto.offset.reset", "latest")
props.put("group.id", "consumer-group")
val consumer: KafkaConsumer[String, String] = new KafkaConsumer[String, String](props)
consumer.subscribe(util.Arrays.asList(topic))
while (true) {
val record = consumer.poll(1000).asScala
for (data <- record.iterator)
println(data.value())
}
}
}
Kafka有办法做到吗?否则,我将不得不发送一个额外的领域从生产者到主题。
1条答案
按热度按时间bvjxkvbb1#
Kafka从v0.10开始提供了一种方法
在这个版本中,所有消息都有一个时间戳信息
data.timestamp
,其中的信息类型由代理上的配置“message.timestamp.type”控制。值应为CreateTime
或者LogAppendTime
.在此版本之前,您必须手动实现它,通常通过修改数据结构来实现。