如何让Kafka Producer代码自动读取更新后的test.json文件

嘿，这个问题我之前也碰到过！核心原因是你的代码大概率只在循环启动前读取了一次test.json的内容，之后就一直用内存里的旧数据跑循环，根本没去检查文件有没有更新。给你几个实用的解决思路：

这是最直接粗暴但有效的办法——把读取文件的逻辑塞进while循环内部，这样每次循环都会重新从磁盘加载最新的test.json内容。示例代码大概是这样：

import json
import time
from kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers='localhost:9092')

while True:
    # 把读取文件的操作移到循环里，每次都读最新的
    try:
        with open('test.json', 'r') as f:
            data = json.load(f)
        # 执行Kafka发送逻辑
        producer.send('your_topic_name', value=json.dumps(data).encode('utf-8'))
    except Exception as e:
        print(f"读取文件或发送消息出错: {e}")
    # 加个短延迟，避免太频繁读取文件占用资源
    time.sleep(1)

小提醒：如果文件体积很大或者循环频率极高，这种方式可能会有性能损耗，更适合文件不大、更新频率不高的场景。

要是不想每次循环都读文件，可以用文件系统监听工具，比如Python的watchdog库——只有当test.json被修改时，才重新读取内容。步骤如下：

1. 先安装依赖：pip install watchdog
2. 编写监听+发送的逻辑框架：

import json
import time
from kafka import KafkaProducer
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

# 用全局变量存最新的文件内容
latest_data = {}
producer = KafkaProducer(bootstrap_servers='localhost:9092')

class FileUpdateHandler(FileSystemEventHandler):
    def on_modified(self, event):
        # 只处理test.json的修改事件
        if not event.is_directory and event.src_path.endswith('test.json'):
            global latest_data
            try:
                with open('test.json', 'r') as f:
                    latest_data = json.load(f)
                print("检测到文件更新，已重新加载数据")
            except Exception as e:
                print(f"更新文件内容出错: {e}")

# 启动文件监听
event_handler = FileUpdateHandler()
observer = Observer()
# 监听当前目录下的文件变化
observer.schedule(event_handler, path='.', recursive=False)
observer.start()

# 持续发送Kafka消息
try:
    while True:
        if latest_data:
            producer.send('your_topic_name', value=json.dumps(latest_data).encode('utf-8'))
        time.sleep(1)
except KeyboardInterrupt:
    # 捕获中断信号，优雅停止监听
    observer.stop()
observer.join()

这种方式更高效，只有文件真的发生变化时才会读取，适合需要长期稳定运行、文件更新频率不确定的场景。

• 一定要给文件读取逻辑加try-except块，避免文件被替换时出现短暂的不存在/权限问题导致程序崩溃。
• 如果是多线程或多进程环境，要注意全局变量的线程安全问题，必要时加锁保护。

内容的提问来源于stack exchange，提问作者Stella