<返回更多

教你用 Python 驾驭 Nacos 配置中心

2023-08-03    AirPython
加入收藏


大家好,我是安果!

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 = 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 使用正常,放到 maclinux 就报错,即 TypeError: cannot pickle '_thread.RLock' object

这里,我们需要重写源码 nacos/cl.NET.py,对非 Windows 系统做一次兼容,改用 RLock 实现

PS:可以发送文末关键字获取源码,直接进行替换

具体可以参考下面 issue

https://github.com/nacos-group/nacos-sdk-python/pull/125

声明:本站部分内容来自互联网,如有版权侵犯或其他问题请与我们联系,我们将立即删除或处理。
▍相关推荐
更多资讯 >>>