在Python中使用nacos的完整方案

pip install nacos-sdk-python

NACOS_SERVER = "http://localhost:8848"  # Nacos服务器地址
NAMESPACE = "public"                   # 命名空间
GROUP = "DEFAULT_GROUP"                # 配置分组
SERVICE_NAME = "python-service"        # 服务名称
DATA_ID = "app-config"                 # 配置标识

from nacos import NacosClient
from nacos_config import *
from fastapi import FastAPI

app = FastAPI()
client = NacosClient(NACOS_SERVER, namespace=NAMESPACE)

@app.on_event("startup")
def register_service():
    client.add_naming_instance(
        service_name=SERVICE_NAME,
        ip="127.0.0.1",
        port=8000,
        metadata={"version": "1.0", "env": "prod"}
    )
    print("服务注册成功")

@app.on_event("shutdown")
def deregister_service():
    client.remove_naming_instance(SERVICE_NAME, "127.0.0.1", 8000)
    print("服务注销完成")

def discover_service():
    instances = client.list_naming_instance(SERVICE_NAME)
    return [{"ip": inst["ip"], "port": inst["port"]} for inst in instances["hosts"]]

# 调用示例
print("当前可用实例:", discover_service())

config = client.get_config(DATA_ID, GROUP)
print("当前配置:", config)

def config_change_callback(config):
    print("配置已更新:", config["content"])
    # 更新应用配置（示例：更新全局变量）
    global app_config
    app_config = config["content"]

# 添加监听
client.add_config_watcher(
    data_id=DATA_ID,
    group=GROUP,
    cb=config_change_callback
)

new_config = {"db_url": "mysql://new_host:3306", "timeout": 30}
client.publish_config(
    data_id=DATA_ID,
    group=GROUP,
    content=str(new_config)
)

import threading
import time

def heartbeat():
    while True:
        client.send_beat(SERVICE_NAME, "127.0.0.1", 8000)
        time.sleep(5)  # 每5秒发送一次心跳

threading.Thread(target=heartbeat, daemon=True).start()

client = NacosClient(
    server_addresses=NACOS_SERVER,
    namespace=NAMESPACE,
    username="nacos",  # 认证用户名
    password="nacos123"  # 认证密码
)

client = NacosClient("192.168.1.10:8848,192.168.1.11:8848")

from tenacity import retry, stop_after_attempt

@retry(stop=stop_after_attempt(3))
def reliable_register():
    client.add_naming_instance(...)

client.publish_config(encrypted_content="K8s#123!@", encrypted=True)

from prometheus_client import Gauge
SERVICE_COUNT = Gauge('nacos_services', 'Number of registered services')
SERVICE_COUNT.set(len(discover_service()))

import logging
logging.basicConfig(level=logging.DEBUG)

pip show nacos-sdk-python
# 确保版本≥0.5.0（支持Nacos 2.x）