from kafka import KafkaConsumer  #导入第三方库
import time
from log.log import logger

#  与kafka建立连接
consumer = KafkaConsumer('kafka-topic',bootstrap_servers=['ip:9092'])
print("连接成功")
for message in consumer:   #开始消费消息
    logger.info("%s value=%s" %(message.topic,message.value))
    print ("%s %s value=%s" % (time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())),message.topic,message.value))

 

转载于:https://www.cnblogs.com/siyz/p/11288402.html

Logo

Kafka开源项目指南提供详尽教程,助开发者掌握其架构、配置和使用,实现高效数据流管理和实时处理。它高性能、可扩展,适合日志收集和实时数据处理,通过持久化保障数据安全,是企业大数据生态系统的核心。

更多推荐