import os
|
from kafka import KafkaConsumer
|
import configparser
|
|
# 配置文件路径
|
config_file = os.path.join(os.getcwd(), 'msg_config.ini')
|
|
# 读取配置文件
|
config = configparser.ConfigParser()
|
config.read(config_file)
|
|
# 从配置文件中获取 Kafka 参数和输出文件路径
|
KAFKA_SERVER = config['KAFKA']['server']
|
TOPIC = config['KAFKA']['topics'].split(', ')
|
output_file = os.path.join(os.getcwd(), config['OUTPUT']['file'])
|
|
print(f"Kafka 服务器地址: {KAFKA_SERVER}")
|
print(f"消费的主题: {TOPIC}")
|
print(f"输出文件路径: {output_file}")
|
|
# 创建 KafkaConsumer 实例
|
consumer = KafkaConsumer(
|
*TOPIC,
|
bootstrap_servers=[KAFKA_SERVER],
|
group_id='test_group',
|
value_deserializer=lambda x: x.decode('utf-8')
|
)
|
|
# 打开文件以追加写入
|
with open(output_file, 'a', encoding='utf-8') as f:
|
for message in consumer:
|
topic = message.topic
|
value = message.value
|
|
f.write(f"Topic: {topic}, Message: {value}\n")
|
f.flush()
|
|
print(f"接收到消息: Topic: {topic}, Message: {value}")
|
