代码收藏家技术教程 2024-12-01

【Python】FastAPI服务注册到Nacos

Python FastAPI服务注册到Nacos

写在开始

一、Nacos的本地安装部署

二、Python服务注册到nacos

Python版本的兼容问题

三、具体实现

写在最后

写在开始

关于Python使用fastapi将服务注册到nacos上的实现，有两种渠道（我只找到这两种，也可能还有其他的），如下。

使用官方的open api接口，会略显笨拙，但是Python能轻松并且简单的实现，在此不做赘述。
使用官方提供的SDK，在官方文档中提供有nacos-sdk-python，可以在环境中安装这个包，然后按照教程使用即可。
不过存在一定的问题，这也是写这篇文章的目的（详情如下）。

一、Nacos的本地安装部署

关于nacos的安装和部署，最简单的肯定就是使用docker了，因为只是做开发测试，所以就在Windows的本地docker上简单安装了一下。具体操作，在nacos官网和csdn上有很多的教程，在此也不赘述了。

二、Python服务注册到nacos

下面就是本文重点了，在开始中，提到了Python服务注册到nacos的两个渠道。下面就是使用nacos提供的sdk去实现服务注册的具体过程中会遇到的第一个问题。

Python版本的兼容问题

如下图，在nacos官网中明确指出，支持Python 2.7、3.6及3.7版本，我是以为这个包只支持3.6或3.7，但是现在Python的大众版本普遍已经3.8、3.9了，甚至已经用上3.12了。
所以我一度认为3.12是不能使用这个nacos-sdk-python包的，就怕出现什么问题。后来open api实在是太笨拙，就抱着试一试的心态，安装了sdk的包。
结果，居然能跑，就很奇怪，也就说明，高版本是可以兼容低版本的。

三、具体实现

关于具体实现的代码，在官网中提供了Flask和Fastapi两种框架的demo，因为我的测试需要fastapi，这里就搬一下对应的代码。
有需要flask demo的，可以自行去官网copy一下，Nacos Python SDK

# 导入FastAPI库，用于构建API服务
from fastapi import FastAPI
from nacos import NacosClient

app = FastAPI()

# 设置Nacos服务器地址，请替换为实际的Nacos服务器地址
SERVER_ADDRESSES = "Your nacos server address"

# 设置Nacos命名空间ID，请替换为实际的命名空间ID
NAMESPACE = "Your nacos namespace"

# 设置Nacos用户名和密码，用于认证访问Nacos服务器
USERNAME = 'Your user name'
PASSWORD = 'Your password'

client = NacosClient(server_addresses=SERVER_ADDRESSES, namespace=NAMESPACE, username=USERNAME, password=PASSWORD,
                     log_level='INFO')


# 定义一个异步函数，在FastAPI应用启动时执行
@app.on_event("startup")
async def startup_event():
    # 在启动时创建一个任务来初始化配置
    asyncio.create_task(init())


# 通过NacosClient获取配置，并存储在应用的状态(state)中，以便后续使用
async def load_config(data_id, group):
    app.state = {'config_data': client.get_config(data_id=data_id, group=group)}


# 异步函数，用于初始化应用所需的各种配置和监听器
async def init():
    data_id = 'test'
    group = 'DEFAULT_GROUP'
    await load_config(data_id, group)

    def on_config_change(cfg):
        # 当Nacos中的配置发生变化时，更新应用状态中的配置数据
        app.state = {'config_data': cfg['content']}

    client.add_config_watcher(data_id, group, cb=on_config_change)
    client.add_naming_instance(service_name='my-flask-service', ip='localhost', port=8000, heartbeat_interval=5)

# 定义一个GET请求的路由，返回简单的欢迎信息及当前从Nacos获取的配置数据
@app.get("/")
def hello_world():
    return f'Hello, World! Config from Nacos: {app.state["config_data"]}'


if __name__ == '__main__':
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8000)

demo都是有注释，就捋一下流程。这段代码通过使用fastapi的event，实现了一个主要的启动方法，startup_event，在调用该方法时，会创建一个异步任务，去执行nacos配置的监控以及Python服务的注册和心跳发送。

并且该方法添加了@app.on_event("startup")的装饰器，该装饰器是fastapi的一个默认启动事件，简单来说就是，在启动这个fastapi demo程序时，会默认先执行startup_event方法里面的代码，这就可以实现在程序启动时就自动注册到nacos了，同时，也存在对应的shoutdown_event可以实现在程序关闭时释放资源。

不过在使用上述demo的过程中，发现了第二个问题。
在fastapi的官方文档中，已经弃用了startup事件，虽然依然能使用，但是会存在资源割裂的情况，虽然也能解决该情况，但是会有点麻烦。于是fastapi官方提供了新的生命周期管理方式lifespan，lifespan作用和前面提到的startup和shutdown一样，都是为了用于解决应用相关生命周期的管理。

所以，在此处，还是选择使用新的生命周期管理方式，新实现了一个对应的demo，如下。

# 导入FastAPI库，用于构建API服务
from fastapi import FastAPI
from nacos import NacosClient
from contextlib import asynccontextmanager

app = FastAPI()

# 设置Nacos服务器地址，请替换为实际的Nacos服务器地址
SERVER_ADDRESSES = "Your nacos server address"

# 设置Nacos命名空间ID，请替换为实际的命名空间ID
NAMESPACE = "Your nacos namespace"

# 设置Nacos用户名和密码，用于认证访问Nacos服务器
USERNAME = 'Your user name'
PASSWORD = 'Your password'

client = NacosClient(server_addresses=SERVER_ADDRESSES, namespace=NAMESPACE, username=USERNAME, password=PASSWORD,
                     log_level='INFO')
                    
# 通过NacosClient获取配置，并存储在应用的状态(state)中，以便后续使用
async def load_config(data_id, group):
    app.state = {'config_data': client.get_config(data_id=data_id, group=group)}


# 异步函数，用于初始化应用所需的各种配置和监听器
async def init():
    data_id = 'test'
    group = 'DEFAULT_GROUP'
    await load_config(data_id, group)

    def on_config_change(cfg):
        # 当Nacos中的配置发生变化时，更新应用状态中的配置数据
        app.state = {'config_data': cfg['content']}

    client.add_config_watcher(data_id, group, cb=on_config_change)
    client.add_naming_instance(service_name='my-flask-service', ip='localhost', port=8000, heartbeat_interval=5)

@asynccontextmanager
async def lifespan(app: FastAPI):
    """
    fastapi生命周期管理，在服务启动时执行初始化操作，并在服务结束时执行清理操作
    Args:
        app:
    Returns:
    """
    # nacos注册
    print('开始注册')
    await init()
    # 进入上下文管理器的上下文。在此之后，FastAPI应用将继续运行
    yield
    print('关闭连接')

与原本的startup_event的实现方法都是一样，只是生命周期不会做割裂，而是通过asynccontextmanager装饰器实现上下文管理。在项目启动时会自动执行nacos的初始化。

写在最后

使用fastapi的lifespan也会出现一点的问题，目前发现的bug就是，在启动程序并开启reload之后，如果中途修改代码，reload机制就会暂时中断程序并重启，此时nacos的心跳任务似乎不会因为主进程的中断而停止，就会报错，如下。

但是，最终的结果并不会影响到程序的reload，在报错之后，程序依然会重启，并且重新注册并发送心跳。
研究了一下之后，似乎可以用信号去控制并关闭nacos的注册。不过目前还没有做到这一步，未完待续…