簡介
python 客戶端
from io import StringIO
from app.customized.nacos_conf.settings import (
SERVER_ADDRESSES,
NAMESPACE,
DATA_ID,
GROUP,
)
from nacos import NacosClient
import configparser
def get_config():
# no auth mode
client = NacosClient(SERVER_ADDRESSES, namespace=NAMESPACE)
# auth mode
# client = nacos.NacosClient(SERVER_ADDRESSES, namespace=NAMESPACE, username="nacos", password="nacos")
# get config
nacos_conf = client.get_config(DATA_ID, GROUP)
parser = configparser.ConfigParser()
like_file = StringIO(nacos_conf)
parser.read_file(like_file)
# parser.get("dev", "ES_SERVER")
return parser
def register_service(**register_info):
# service_name = register_info.get("service_name")
# ip = register_info.get("ip")
# port = register_info.get("port")
# cluster_name = register_info.get("cluster_name")
# weight = register_info.get("weight")
# metadata = register_info.get("metadata")
# enable = register_info.get("enable")
# healthy = register_info.get("healthy")
client = NacosClient(SERVER_ADDRESSES, namespace=NAMESPACE)
client.add_naming_instance(**register_info)
def nacos_watcher():
watch_info = "watcher"
print(watch_info)
return watch_info
def send_heartbeat():
client = NacosClient(SERVER_ADDRESSES, namespace=NAMESPACE)
res = client.send_heartbeat(
"test.pole-recommender", "192.168.60.86", 5000, "testCluster2", 0.1, "{}"
)["clientBeatInterval"]
print(res)
if __name__ == "__main__":
get_config()
# nacos.NacosClient.add_config_watchers(DATA_ID, GROUP, nacos_watcher)
register_info = dict(
service_name="test.pole-recommender",
ip="192.168.60.86",
port=5000,
cluster_name="testCluster2",
weight=1.0,
metadata="{}",
enable=True,
healthy=True,
ephemeral=False,
)
register_service(**register_info)
send_heartbeat()
# content='publish info'
# timeout = 60
# nacos.NacosClient.publish_config(DATA_ID, GROUP, content, timeout)
# nacos.NacosClient.set_debugging()
# only effective within the current process
