关于python web项目使用nacos
好多年没来写原创了，最近自己写了一个python web 要使用nacos做微服务统一配置管理和服务注册与发现还有负载均衡这些。本来可以用官网推荐的那个nacos sdk 但项目上有一些很变态的要求，于是就自己写了一个。
这个sdk 支持加载任意命名空间下 任意组的 json配置文件，且可以加载任意数量个的配置文件，你想加载几个加载几个。
服务注册时候可以同时注册不同的注册中心（很变态的要求）
服务发现消费者端，允许消费者跨组 跨命名空间 自由调用。这点spring cloud nacos 是只允许同命名空间下同组的服务进行调用的。
项目我开源到git上了，地址： https://github.com/KcangYan/nacos-python-sdk
也打包发布到pypi上了可以使用pip下载安装
pip install KcangNacos
实现功能
- 实现python web app 作为客户端使用nacos远程配置中心的功能(目前仅支持json格式的配置文件)，实现实时监听配置，及时同步远程配置。并且运行nacos跨组别调用配置，以及同一个web加载多个配置
- 实现python web app 作为服务实例，注册进nacos注册管理中心的功能
- 实现注解式的无侵入的python nacos消费者客户端，且带有负载均衡的能力，且 可以实现nacos跨组调用服务。
难得想写一篇文章贡献一下，不多废话直接上代码。
使用方法
首先通过pip安装 pip install KcangNacos
然后在项目中导入类
import KcangNacos.nacos as nacos
创建初始nacos连接对象
nacosServer = nacos.nacos(ip=nacosIp,port=nacosPort)
nacos远程配置中心
nacosServer.config(dataId="demo-python.json",group="dev",tenant="python",myConfig=myConfig.GlobalConfig)
执行以上代码即可启动加载远程配置中心，并实时监听配置中心变化，及时变更本地配置 。
其中tenant=namespaceId，目前只支持 nacos config json格式的配置
myConfig.GlobalConfig为用户本地的配置类对象，必须是dict。
启动后会根据远程配置中心的json数据按键值装配到已导入的本地配置类里。
nacos服务注册
nacosServer.registerService(serviceIp=myConfig.ip,servicePort=myConfig.port,serviceName="python-provider",namespaceId="python",groupName="dev")
执行以上代码即可启动nacos服务注册，会将实例注册进nacos注册中心。
nacos进程健康检查
nacosServer.healthyCheck()
执行以上代码即可开启本地naocs进程的健康检查
考虑到本地的服务注册和远程配置中心监听进程的安全，怕他万一挂掉，所以设置这个功能
该服务会检查naocs进程是否健康，如果挂掉了会自动把进程重启
怕影响本地服务性能的可以不开，但建议还是开着。
那如果万一健康检查线程也挂了怎么办？可以利用nacos对象里的healthy属性获得当前健康进程的秒级时间戳
可以利用 int(time.time()) – nacosServer.healthy 获得健康检查进程的执行时间，时间过长则认为挂掉了
再执行一次这行代码就可以重新启动检查检查进程
nacos消费者
nacos的跨组调用服务客户端，实现注解式无侵入的功能实现，且带有负载均衡的能力。
具体如下：（给了四种demo，大概基本上的场景都覆盖了）
nacosClient = nacos.nacosBalanceClient(ip=nacosIp, port=nacosPort,
serviceName="python-provider",
group="dev", namespaceId="python")
建立nacosBalanceClient对象，使用注解
@nacosClient.customRequestClient(method="GET", url="/api/test1")
def apiTest1():
pass
@nacosClient.customRequestClient(method="GET", url="/api/test2")
def apiTest2(id1: int, id2: int):
pass
@nacosClient.customRequestClient(method="POST", url="/api/test3")
def apiTest3(formData):
pass
@nacosClient.customRequestClient(method="POST", url="/api/test4", requestParamJson=True)
def apiTest4(jsonData):
pass
@nacosClient.customRequestClient(method="GET", url="/api/test5")
def apiTest5(*args,**kwargs):
pass
在路由方法中直接调用注解了的接口函数即可
@app.route(Router + "/test1", methods=['GET'])
def consumerTest1():
return consumerDemo.apiTest1()
customRequestClient方法自带负载均衡
客户端自带了一个 timeout 和 error fallback 函数，并且默认超时时间是3秒，如果你想自定义
def errorFun():
return "自定义错误函数"
或者
def errorFun(*args,**kwargs):
for item in args:
print(item)
return "自定义错误"
nacosClient = nacos.nacosBalanceClient(ip=nacosIp, port=nacosPort,
serviceName="python-provider",
group="dev", namespaceId="python",timeout=3,
timeOutFun=errorFun,fallbackFun=errorFun)
设置timeout为服务超时时间。
结尾
我在git上放了一个flask的使用demo，有需要的可以去下载，使用起来并不难。
https://github.com/KcangYan/nacos-python-sdk
如果需要熔断器和限流器的话可以看我的另一篇博客
Original: https://blog.csdn.net/Yezeqi0328/article/details/115803747
Author: Kcang
Title: python微服务，使用nacos，实现注册中心，服务注册，加载多个配置文件，负载均衡客户端以及跨组调用服务
原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/735276/
转载文章受原作者版权保护。转载请注明原作者出处！