QwenModel/text_analysis/tools/zk_util.py

'''
     监听数据
    "{
    "scenes_id":2222,
    "operation":"stop",
    "version":5
}"
scenes_id=2222
version!=0

'''
from kazoo.client import KazooClient
from kazoo.protocol.states import EventType
import time

# 连接到ZooKeeper服务器
zk = KazooClient(hosts='172.18.1.146:2181,172.18.1.147:2181,172.18.1.148:2181')
zk.start()

# 定义数据变更时的回调函数
def data_change_listener(event):
    if event.type == EventType.CHANGED:
        data, stat = zk.get("/analyze")
        print("Data changed on node /analyze: {data.decode('utf-8')}")
    elif event.type == EventType.DELETED:
        print("Node /analyze has been deleted")

# 设置监听器
@zk.DataWatch("/analyze")
def watch_node(data, stat, event):
    if event is not None:
        data_change_listener(event)

# 保持程序运行以监听节点变化
try:
    while True:
        print("ok")
        time.sleep(1)
except KeyboardInterrupt:
    print("Stopping...")

# 关闭连接
zk.stop()
zk.close()