教你用 python 駕馭 Nacos 配置中心

作者：星安果 2023-08-03 08:51:07
云計算
云原生 Nacos 是阿里巴巴開源的項目，用于構建云原生應用的動態(tài)服務發(fā)現(xiàn)、配置管理和服務管理平臺，核心特征包含：服務發(fā)現(xiàn)、服務健康監(jiān)測、動態(tài)配置服務、動態(tài) DNS 服務、服務及其元數(shù)據(jù)管理。

創(chuàng)新互聯(lián)專業(yè)為企業(yè)提供林甸網(wǎng)站建設、林甸做網(wǎng)站、林甸網(wǎng)站設計、林甸網(wǎng)站制作等企業(yè)網(wǎng)站建設、網(wǎng)頁設計與制作、林甸企業(yè)網(wǎng)站模板建站服務，10多年林甸做網(wǎng)站經(jīng)驗，不只是建網(wǎng)站，更提供有價值的思路和整體網(wǎng)絡服務。

大家好，我是安果！

Nacos 是阿里巴巴開源的項目，用于構建云原生應用的動態(tài)服務發(fā)現(xiàn)、配置管理和服務管理平臺

核心特征包含：服務發(fā)現(xiàn)、服務健康監(jiān)測、動態(tài)配置服務、動態(tài) DNS 服務、服務及其元數(shù)據(jù)管理

如果想在 Python 項目中利用 Nacos 動態(tài)配置服務該怎么做呢？

1、安裝依賴

nacos-sdk-python 項目是 Nacos OpenAPI 的 Python 實現(xiàn)，可用于監(jiān)聽 Nacos 配置文件的數(shù)據(jù)變動

# 安裝以來
pip3 install nacos-sdk-python

# Nacos配置文件為yaml的依賴
pip3 install pyyaml

項目地址：https://github.com/nacos-group/nacos-sdk-python

2、基礎使用（yaml）

以 YAML 配置文件為例

首先，通過 Nacos 連接信息（連接信息、命名空間、用戶名及密碼）創(chuàng)建一個 Nacos 客戶端連接對象

import nacos

# 連接地址
SERVER_ADDRESSES = "192.*.*.*"
SERVER_PORT = '8848'

# 命名空間
NAMESPACE = "public"

# 賬號信息
USERNAME = 'nacos'
PASSWORD = 'nacos'

# 創(chuàng)建一個連接對象
client = nacos.NacosClient(f'{SERVER_ADDRESSES}:{SERVER_PORT}', namespace=NAMESPACE, username=USERNAME,
                           password=PASSWORD

然后，通過分組名和服務 ID 解析出某個服務的配置，以 YAML 的形式進行數(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)

    # 通過鍵路徑，解析出數(shù)據(jù)
    result = config_data['arg1']['arg2']

    print(result)

# 服務id（鍵）
data_id = "service_name"

# 分組名稱，默認為：DEFAULT_GROUP
group = "DEFAULT_GROUP"

# 初始化解析
init(data_id, group)

最后，通過分組名和服務 ID，使用 Nacos 連接信息添加一個監(jiān)聽事件，這樣當 Nacos 配置變動時，程序能及時獲取變動后的數(shù)據(jù)

# Nacos數(shù)據(jù)變動時觸發(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)聽Nacos數(shù)據(jù)變動
def add_nacos_listener(data_id, group):
    client.add_config_watcher(data_id=data_id, group=group, cb=nacos_data_change_callback)

# 添加監(jiān)聽事件
add_nacos_listener(data_id, group)

3、properties 文件

與 YAML 配置文件的區(qū)別是

• YAML 使用縮進和冒號來表示層次結構
• Properties 使用等號連接鍵值對

在監(jiān)聽 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數(shù)據(jù)變動時觸發(fā)
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數(shù)據(jù)變動了，address：", address)

4、Python Web + Nacos

在 Python Web 應用中，如果想結合 Nacos 的動態(tài)配置，需要按下面步驟進行

這里以 FastAPI 為例進行講解

首先，定義 Nacos 客戶端連接對象及一個全局變量

PS：全局變量用于測試演示

import nacos

client = nacos.NacosClient(f'{SERVER_ADDRESSES}:{SERVER_PORT}', namespace=NAMESPACE, username=USERNAME,
                           password=PASSWORD)

# 定義一個全局變量
arg1 = ''

然后，實例化一個 FastAPI 對象，并在應用啟動時利用 asyncio 創(chuàng)建一個監(jiān)聽事件

from fastapi import FastAPI
import nacos
import uvicorn
import asyncio

app = FastAPI()

# 運行時觸發(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)聽事件中，通過分組名和服務 ID 進行初始化及數(shù)據(jù)監(jiān)聽

# 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數(shù)據(jù)變動時觸發(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)聽，用于數(shù)據(jù)變動監(jiān)聽
    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 配置文件數(shù)據(jù)變化時，通過接口都能實時獲取最新的數(shù)據(jù)

5、防坑

如 nacos-sdk-python 項目介紹，作者最高只對 Python3.7 及 Nacos 1.3.2做了兼容

在實際測試過程中，發(fā)現(xiàn)程序在 Windows 使用正常，放到 Mac 或 Linux 就報錯，即 TypeError: cannot pickle '_thread.RLock' object

這里，我們需要重寫源碼 nacos/clinet.py，對非 Windows 系統(tǒng)做一次兼容，改用 RLock 實現(xiàn)

PS：可以發(fā)送文末關鍵字獲取源碼，直接進行替換

具體可以參考下面 issue

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

本文標題：教你用Python駕馭Nacos配置中心
文章路徑：http://www.5511xx.com/article/coeehss.html