关于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

实现功能

1. 实现python web app 作为客户端使用nacos远程配置中心的功能(目前仅支持json格式的配置文件)，实现实时监听配置，及时同步远程配置。并且运行nacos跨组别调用配置，以及同一个web加载多个配置
2. 实现python web app 作为服务实例，注册进nacos注册管理中心的功能
3. 实现注解式的无侵入的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(): #无参数GET接口
      pass
      
  @nacosClient.customRequestClient(method="GET", url="/api/test2")
  def apiTest2(id1: int, id2: int): #带参数GET RESTFUL接口
      pass

  @nacosClient.customRequestClient(method="POST", url="/api/test3")
  def apiTest3(formData): #POST 传formData格式数据接口，传入一个dict对象
      pass

  @nacosClient.customRequestClient(method="POST", url="/api/test4", requestParamJson=True)
  def apiTest4(jsonData): #POST 传json格式数据接口，传入一个dict对象，内置了对象转str
      pass  

  @nacosClient.customRequestClient(method="GET", url="/api/test5")
  def apiTest5(*args,**kwargs): #对于非RESTFUL的get请求也可以使用
      pass    
  #传入键值对即可 如  传入   apiTest5(x=1,y=2) 即  /api/test5?x=1&y=2      

  在路由方法中直接调用注解了的接口函数即可 
  @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

哦对了，整个项目中使用了python自带的logging模块打印系统日志。
如果需要熔断器和限流器的话可以看我的另一篇博客