import json
|
import time
|
import configparser
|
from kafka import KafkaProducer
|
from threading import Timer
|
|
# 读取配置文件
|
def load_config(config_file):
|
config = configparser.ConfigParser()
|
config.read(config_file)
|
return config
|
|
# 读取 JSON 文件内容
|
def load_json(json_file):
|
with open(json_file, 'r') as f:
|
return json.load(f)
|
|
# 发送消息到 Kafka
|
def send_message(producer, topic, message):
|
producer.send(topic, value=json.dumps(message).encode('utf-8'))
|
print(f"Sent message to {topic}: {message}")
|
|
# 定时发送消息
|
def send_periodically(producer, topic, message, interval, stop_flag):
|
if stop_flag[0]: # 如果 stop_flag 是 True，则停止发送
|
send_message(producer, topic, message)
|
# 设置定时器，按照给定的间隔继续发送
|
Timer(interval, send_periodically, [producer, topic, message, interval, stop_flag]).start()
|
else:
|
print("Stopping periodic message sending.")
|
|
# 停止定时器
|
def stop_sending(stop_flag):
|
stop_flag[0] = False
|
print("Message sending stopped.")
|
|
def main():
|
# 配置文件路径
|
config_file = 'msg_config.ini'
|
config = load_config(config_file)
|
|
# Kafka 配置
|
kafka_server = config['KAFKA']['server']
|
topics = config['KAFKA']['topics'].split(',') # topics 列表
|
send_topics = config['SEND_MSG']['send_topics']
|
run_send = config.getboolean('SEND_MSG', 'run_send') # run_send 为布尔值
|
send_interval = config.getint('SEND_MSG', 'send_interval')
|
json_file = config['SEND_MSG']['json_file']
|
|
# 读取消息内容
|
message = load_json(json_file)
|
|
# 创建 Kafka producer
|
producer = KafkaProducer(bootstrap_servers=kafka_server)
|
|
# 如果需要发送消息，则启动定时任务
|
if run_send:
|
print(f"Sending messages to {send_topics} every {send_interval} seconds...")
|
|
# 使用列表来实现可修改的标志位
|
stop_flag = [True]
|
|
# 启动定时器
|
send_periodically(producer, send_topics, message, send_interval, stop_flag)
|
|
# 例如，可以在10秒后停止定时发送
|
#time.sleep(10)
|
#stop_sending(stop_flag)
|
else:
|
print("run_send is false. No messages will be sent.")
|
|
if __name__ == "__main__":
|
main()
|
