江苏龙网

  1. 主页 > 生活百科 > >

教你用 Python 驾驭 Nacos 配置中心

Python


教你用 Python 驾驭 Nacos 配置中心

文章插图
大家好，我是安果！
Nacos 是阿里巴巴开源的项目，用于构建云原生应用的动态服务发现、配置管理和服务管理平台
核心特征包含：服务发现、服务健康监测、动态配置服务、动态 DNS 服务、服务及其元数据管理
如果想在 Python/ target=_blank class=infotextkey>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 = https://www.isolves.com/it/cxkf/yy/Python/2023-08-03/yaml.load(config, Loader=yaml.FullLoader)# 通过键路径，解析出数据result = config_data['arg1']['arg2']print(result)# 服务id（键）data_id = "service_name"# 分组名称，默认为：DEFAULT_GROUPgroup = "DEFAULT_GROUP"# 初始化解析init(data_id, group)最后，通过分组名和服务 ID，使用 Nacos 连接信息添加一个监听事件，这样当 Nacos 配置变动时，程序能及时获取变动后的数据
# Nacos数据变动时触发def nacos_data_change_callback(config):# 数据解析nacos_data = https://www.isolves.com/it/cxkf/yy/Python/2023-08-03/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：全局变量用于测试演示
import nacosclient = nacos.NacosClient(f'{SERVER_ADDRESSES}:{SERVER_PORT}', namespace=NAMESPACE, username=USERNAME,password=PASSWORD)# 定义一个全局变量arg1 = ''然后，实例化一个 FastAPI 对象，并在应用启动时利用 asyncio 创建一个监听事件
from fastapi import FastAPIimport nacosimport uvicornimport asyncioApp = 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 arg1config_list = config['content'].split("n")...# 配置的地址arg1 = properties['address']print("arg1:", arg1)async def event_listener():data_id = "service_name"group = "DEFAULT_GROUP"# 初始化awaitinit(data_id, group)# Nacos配置监听，用于数据变动监听client.add_config_watcher(data_id=data_id, group=group, cb=nacos_data_change_callback)


推荐阅读


上一篇：关于自动限流的思考

下一篇：SpringBoot中使用Cache提升接口性能详解