大家好，我是安果！
Nacos 是阿里巴巴開(kāi)源的項(xiàng)目，用于構(gòu)建云原生應(yīng)用的動(dòng)態(tài)服務(wù)發(fā)現(xiàn)、配置管理和服務(wù)管理平臺(tái)
核心特征包含：服務(wù)發(fā)現(xiàn)、服務(wù)健康監(jiān)測(cè)、動(dòng)態(tài)配置服務(wù)、動(dòng)態(tài) DNS 服務(wù)、服務(wù)及其元數(shù)據(jù)管理
如果想在 Python 項(xiàng)目中利用 Nacos 動(dòng)態(tài)配置服務(wù)該怎么做呢？
1、安裝依賴
nacos-sdk-python 項(xiàng)目是 Nacos OpenAPI 的 Python 實(shí)現(xiàn)，可用于監(jiān)聽(tīng) Nacos 配置文件的數(shù)據(jù)變動(dòng)
# 安裝以來(lái)
pip3 install nacos-sdk-python
# Nacos配置文件為yaml的依賴
pip3 install pyyaml
項(xiàng)目地址：
https://github.com/nacos-group/nacos-sdk-python
2、基礎(chǔ)使用（yaml）
以 YAML 配置文件為例
首先，通過(guò) Nacos 連接信息（連接信息、命名空間、用戶名及密碼）創(chuàng)建一個(gè) Nacos 客戶端連接對(duì)象
import nacos
# 連接地址
SERVER_ADDRESSES = "192.*.*.*"
SERVER_PORT = '8848'
# 命名空間
NAMESPACE = "public"
# 賬號(hào)信息
USERNAME = 'nacos'
PASSWORD = 'nacos'
# 創(chuàng)建一個(gè)連接對(duì)象
client = nacos.NacosClient(f'{SERVER_ADDRESSES}:{SERVER_PORT}', namespace=NAMESPACE, username=USERNAME,
password=PASSWORD
然后，通過(guò)分組名和服務(wù) ID 解析出某個(gè)服務(wù)的配置，以 YAML 的形式進(jìn)行數(shù)據(jù)解析
import yaml
# 初始化
def init(data_id, group):
config = client.get_config(data_id, group)
# 配置數(shù)據(jù)解析（YAML）
config_data = yaml.load(config, Loader=yaml.FullLoader)
# 通過(guò)鍵路徑，解析出數(shù)據(jù)
result = config_data['arg1']['arg2']
print(result)
# 服務(wù)id（鍵）
data_id = "service_name"
# 分組名稱，默認(rèn)為：DEFAULT_GROUP
group = "DEFAULT_GROUP"
# 初始化解析
init(data_id, group)
最后，通過(guò)分組名和服務(wù) ID，使用 Nacos 連接信息添加一個(gè)監(jiān)聽(tīng)事件，這樣當(dāng) Nacos 配置變動(dòng)時(shí)，程序能及時(shí)獲取變動(dòng)后的數(shù)據(jù)
# Nacos數(shù)據(jù)變動(dòng)時(shí)觸發(fā)
def nacos_data_change_callback(config):
# 數(shù)據(jù)解析
nacos_data = yaml.load(config['content'], Loader=yaml.FullLoader)
# 讀取鍵值
result = nacos_data['arg1']['arg2']
print(result)
# 監(jiān)聽(tīng)Nacos數(shù)據(jù)變動(dòng)
def add_nacos_listener(data_id, group):
client.add_config_watcher(data_id=data_id, group=group, cb=nacos_data_change_callback)
# 添加監(jiān)聽(tīng)事件
add_nacos_listener(data_id, group)
3、properties 文件
與 YAML 配置文件的區(qū)別是
在監(jiān)聽(tīng) Nacos 配置文件這一功能上，我們只需要修改解析的邏輯即可
import nacos
# 解析Properties配置文件（Nacos）
# 初始化
def init(data_id, group):
# 換行符進(jìn)行分割，存入列表中
config_list = client.get_config(data_id, group).split("\n")
properties = {}
for config_item in config_list:
# 過(guò)濾有用的鍵值對(duì)
if config_item.find('=') > 0:
strs = config_item.replace('\n', '').split('=')
properties[strs[0]] = strs[1]
# 配置的地址
address = properties['address']
print(address)
# Nacos數(shù)據(jù)變動(dòng)時(shí)觸發(fā)
def nacos_data_change_callback(config):
config_list = config['content'].split("\n")
properties = {}
for config_item in config_list:
# 過(guò)濾有用部分
if config_item.find('=') > 0:
strs = config_item.replace('\n', '').split('=')
properties[strs[0]] = strs[1]
# 配置的地址
address = properties['address']
print("Nacos數(shù)據(jù)變動(dòng)了，address：", address)
4、Python Web + Nacos
在 Python Web 應(yīng)用中，如果想結(jié)合 Nacos 的動(dòng)態(tài)配置，需要按下面步驟進(jìn)行
這里以 FastAPI 為例進(jìn)行講解
首先，定義 Nacos 客戶端連接對(duì)象及一個(gè)全局變量
PS：全局變量用于測(cè)試演示
import nacos
client = nacos.NacosClient(f'{SERVER_ADDRESSES}:{SERVER_PORT}', namespace=NAMESPACE, username=USERNAME,
password=PASSWORD)
# 定義一個(gè)全局變量
arg1 = ''
然后，實(shí)例化一個(gè) FastAPI 對(duì)象，并在應(yīng)用啟動(dòng)時(shí)利用 asyncio 創(chuàng)建一個(gè)監(jiān)聽(tīng)事件
from fastapi import FastAPI
import nacos
import uvicorn
import asyncio
app = FastAPI()
# 運(yùn)行時(shí)觸發(fā)
@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)
在監(jiān)聽(tīng)事件中，通過(guò)分組名和服務(wù) ID 進(jìn)行初始化及數(shù)據(jù)監(jiān)聽(tīng)
# Nacos初始化
async def init(data_id, group):
global arg1
# 換行符進(jìn)行分割，存入列表中
config_list = client.get_config(data_id, group).split("\n")
...
# 配置的地址
arg1 = properties['address']
print("arg1:", arg1)
# Nacos數(shù)據(jù)變動(dòng)時(shí)觸發(fā)
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配置監(jiān)聽(tīng)，用于數(shù)據(jù)變動(dòng)監(jiān)聽(tīng)
client.add_config_watcher(data_id=data_id, group=group, cb=nacos_data_change_callback)
最后，定義一個(gè)簡(jiǎn)單的接口用于獲取變量的值
# 定義一個(gè)全局變量
arg1 = ''
@app.get("/")
async def index():
global arg1
return {"message": arg1}
這樣，當(dāng) Nacos 配置文件數(shù)據(jù)變化時(shí)，通過(guò)接口都能實(shí)時(shí)獲取最新的數(shù)據(jù)
5、防坑
如 nacos-sdk-python 項(xiàng)目介紹，作者最高只對(duì) Python3.7 及 Nacos 1.3.2做了兼容
在實(shí)際測(cè)試過(guò)程中，發(fā)現(xiàn)程序在 Windows 使用正常，放到 Mac 或 Linux 就報(bào)錯(cuò)，即 TypeError: cannot pickle '_thread.RLock' object
這里，我們需要重寫(xiě)源碼 nacos/clinet.py，對(duì)非 Windows 系統(tǒng)做一次兼容，改用 RLock 實(shí)現(xiàn)
PS：可以發(fā)送文末關(guān)鍵字獲取源碼，直接進(jìn)行替換
具體可以參考下面 issue
https://github.com/nacos-group/nacos-sdk-python/pull/125
