大家好，我是安果！
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 配置文件的区别是
YAML 使用缩进和冒号来表示层次结构
Properties 使用等号连接键值对
在监听 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：全局变量用于测试演示
然后，实例化一个 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/clinet.py，对非 Windows 系统做一次兼容，改用 RLock 实现
PS：可以发送文末关键字获取源码，直接进行替换
具体可以参考下面 issue
https://github.com/nacos-group/nacos-sdk-python/pull/125
我已经将文中所有源码上传到后台，回复关键字「 230803 」即可以获取完整源码
如果你觉得文章还不错，请大家 点赞、分享、留言 下，因为这将是我持续输出更多优质文章的最强动力！
推荐阅读
用 Python 远程控制 Windows 服务器，太好用了！
JavaScript 逆向爬虫中的浏览器调试常见技巧
JavaScript 逆向爬虫中的浏览器调试常见技巧（下）
破解响应加密：Python 与 JS 逆向结合的最佳实践
END
好文和朋友一起看~
