来源： AIrPython/ target=_blank class=infotextkey>Python
作者：星安果
Nacos 是阿里巴巴开源的项目，用于构建云原生应用的动态服务发现、配置管理和服务管理平台
核心特征包含：服务发现、服务健康监测、动态配置服务、动态 DNS 服务、服务及其元数据管理
如果想在 Python 项目中利用 Nacos 动态配置服务该怎么做呢？
1、安装依赖
nacos-sdk-python 项目是 Nacos OpenAPI 的 Python 实现，可用于监听 Nacos 配置文件的数据变动
# 安装依赖
pip3 install nacos-sdk-python
# Nacos配置文件为yaml的依赖
pip3 install pyyaml
项目地址：
https://Github.com/nacos-group/nacos-sdk-python
2、基础使用（yaml）
以 YAML 配置文件为例
首先，通过 Nacos 连接信息（连接信息、命名空间、用户名及密码）创建一个 Nacos 客户端连接对象
import nacos
# 连接地址
SERVER_ADDRESSES = "192.*.*.*"
SERVER_PORT = '8848'
# 命名空间
NAMESPACE = "public"
# 账号信息
USERNAME = 'nacos'
PASSword = 'nacos'
# 创建一个连接对象
client = nacos.NacosClient(f'{SERVER_ADDRESSES}:{SERVER_PORT}', namespace=NAMESPACE, username=USERNAME,
password=PASSWORD
然后，通过分组名和服务 ID 解析出某个服务的配置，以 YAML 的形式进行数据解析
import yaml
# 初始化
def init(data_id, group):
config = client.get_config(data_id, group)
# 配置数据解析（YAML）
config_data = yaml.load(config, Loader=yaml.FullLoader)
# 通过键路径，解析出数据
result = config_data['arg1']['arg2']
print(result)
# 服务id（键）
data_id = "service_name"
# 分组名称，默认为：DEFAULT_GROUP
group = "DEFAULT_GROUP"
# 初始化解析
init(data_id, group)
最后，通过分组名和服务 ID，使用 Nacos 连接信息添加一个监听事件，这样当 Nacos 配置变动时，程序能及时获取变动后的数据
# Nacos数据变动时触发
def nacos_data_change_callback(config):
# 数据解析
nacos_data = yaml.load(config['content'], Loader=yaml.FullLoader)
# 读取键值
result = nacos_data['arg1']['arg2']
print(result)
# 监听Nacos数据变动
def add_nacos_listener(data_id, group):
client.add_config_watcher(data_id=data_id, group=group, cb=nacos_data_change_callback)
# 添加监听事件
add_nacos_listener(data_id, group)
3、properties 文件
与 YAML 配置文件的区别是
在监听 Nacos 配置文件这一功能上，我们只需要修改解析的逻辑即可
import nacos
# 解析Properties配置文件（Nacos）
# 初始化
def init(data_id, group):
# 换行符进行分割，存入列表中
config_list = client.get_config(data_id, group).split("n")
properties = {}
for config_item in config_list:
# 过滤有用的键值对
if config_item.find('=') > 0:
strs = config_item.replace('n', '').split('=')
properties[strs[0]] = strs[1]
# 配置的地址
address = properties['address']
print(address)
# Nacos数据变动时触发
def nacos_data_change_callback(config):
config_list = config['content'].split("n")
properties = {}
for config_item in config_list:
# 过滤有用部分
if config_item.find('=') > 0:
strs = config_item.replace('n', '').split('=')
properties[strs[0]] = strs[1]
# 配置的地址
address = properties['address']
print("Nacos数据变动了，address：", address)
4、Python Web + Nacos
在 Python Web 应用中，如果想结合 Nacos 的动态配置，需要按下面步骤进行
这里以 FastAPI 为例进行讲解
首先，定义 Nacos 客户端连接对象及一个全局变量
PS：全局变量用于测试演示
import nacos
client = nacos.NacosClient(f'{SERVER_ADDRESSES}:{SERVER_PORT}', namespace=NAMESPACE, username=USERNAME,
password=PASSWORD)
# 定义一个全局变量
arg1 = ''
然后，实例化一个 FastAPI 对象，并在应用启动时利用 asyncio 创建一个监听事件
from fastapi import FastAPI
import nacos
import uvicorn
import asyncio
App = FastAPI()
# 运行时触发
@app.on_event("startup")
async def startup_event():
asyncio.create_task(event_listener())
if __name__ == '__main__':
uvicorn.run("demo_fastapi:app", host="0.0.0.0", port=8000, reload=True)
在监听事件中，通过分组名和服务 ID 进行初始化及数据监听
# Nacos初始化
async def init(data_id, group):
global arg1
# 换行符进行分割，存入列表中
config_list = client.get_config(data_id, group).split("n")
...
# 配置的地址
arg1 = properties['address']
print("arg1:", arg1)
# Nacos数据变动时触发
def nacos_data_change_callback(config):
global arg1
config_list = config['content'].split("n")
...
# 配置的地址
arg1 = properties['address']
print("arg1:", arg1)
async def event_listener():
data_id = "service_name"
group = "DEFAULT_GROUP"
# 初始化
await init(data_id, group)
# Nacos配置监听，用于数据变动监听
client.add_config_watcher(data_id=data_id, group=group, cb=nacos_data_change_callback)
最后，定义一个简单的接口用于获取变量的值
# 定义一个全局变量
arg1 = ''
@app.get("/")
async def index():
global arg1
return {"message": arg1}
这样，当 Nacos 配置文件数据变化时，通过接口都能实时获取最新的数据
5、防坑
如 nacos-sdk-python 项目介绍，作者最高只对 Python3.7 及 Nacos 1.3.2做了兼容
在实际测试过程中，发现程序在 windows 使用正常，放到 mac 或 linux 就报错，即 TypeError: cannot pickle '_thread.RLock' object
这里，我们需要重写源码 nacos/cl.NET.py，对非 Windows 系统做一次兼容，改用 RLock 实现
