根据项目需要 将python服务也纳入Nacos 中进行统一管理，所以进行python Nacos 项目适配。

记录本此适配过程。

python 安装不在说明。

系统版本：Linux 5.4.18-87.76-generic KYLINOS SMP Thu Aug 31 09:05:44 UTC 2023 aarch64 aarch64 aarch64 GNU/Linux

python 版本 ：3.11.2

nacos-sdk-python :1.0.0

Nacos:2.3.0

1.安装 nacos-sdk-python

      pip  install   nacos-sdk-python

2. Nacos 注册

 NacosService.py

import nacos


class NacosService:

    def __init__(self,config):
        print(config)
        self.server_addresses = config["nacos_addr"]
        self.namespace =   config["namespace"]
        self.serviceName = config["serviceName"]
        self.ip =   config["ip"]
        self.port =   int(config["port"])
    def connect_service_center(self):
        self.client = nacos.NacosClient(self.server_addresses, namespace=self.namespace)
        print("Nacos Service Register  Start !")
        
    def register_instance(self):
        self.client.add_naming_instance(self.serviceName, self.ip, port=self.port)
        print("Nacos Service Register  Success !")

    def deregister_instance(self):
        self.client.remove_naming_instance(self.serviceName, self.ip, port=self.port)
        print("Nacos Service DeregisterInstance  Success !")
    
    def send_heartbeat(self):
        self.client.send_heartbeat(self.serviceName, self.ip, port=self.port)

3.集成 Flask 进行测试

TestNacos.py

from flask import Flask, request
import json
import uuid
import os
import threading
import time

import NacosManager
from utils.config import ConfigService

app = Flask(__name__)

@app.route('/hello', methods=['GET'])
def hello():
    if request.method == 'GET':
        code =200
        return_result = {'code': code, 'message': '处理成功', 'data': "hello"}
        return json.dumps(return_result, ensure_ascii=False)
        
  

def initNacos(serverConfig):
    nacosService= NacosManager.NacosService(serverConfig)
    nacosService.connect_service_center()
    nacosService.register_instance()
    while True:
        try:
            nacosService.send_heartbeat()
            time.sleep(20)
        except Exception as e:
            print(e)
            time.sleep(5)  
        
if __name__ == "__main__":
    serverConfig =ConfigService.readConfig()
    nacos_thread =threading.Thread(target=initNacos,args=(serverConfig,),daemon =True)
    nacos_thread.start()
    app.run(host="0.0.0.0", port=int(serverConfig["port"]), debug=True)

Nacos 因为需要一个线程专门来作为监听，所以开了一个线程在程序运行中 一直保持和注册中心通信，每20秒发送心跳。

3.读取配置

ConfigService.py

import json
import os

class ConfigService: 

    def readConfig():
        print(os.getcwd())
        filePath=os.path.join("config","Nacos.json")
        with open(filePath) as json_file:
            config = json.load(json_file)
        return  config   

Nacos.json

{
	"nacos_addr":"127.0.0.1:8848",
 "namespace":"Test-Nacos",
	"serviceName":"Test-Nacos-py"	,
	"clusterName":"DefaultCluster",
	"ip":"127.0.0.1",
	"port":9588
}

效果