如何使用 Nacos 对 Python 服务进行服务发现与注册|江阴雨辰互联

如何使用 Nacos 对 Python 服务进行服务发现与注册

在微服务架构中，服务发现与注册是实现服务间通信的关键环节。Nacos 作为一款强大的服务发现与配置管理工具，提供了简单易用的接口来支持 Python 服务的注册与发现。以下将详细介绍如何在 Python 项目中使用 Nacos 实现服务的注册与发现。

最近由于产品架构在考虑是否需要上微服务,于是做个nacos对python服务进行服务注册与发现的案例来向开发团队进行演示,所以记录一下这次测试

一、Nacos 简介

Nacos 是阿里巴巴开源的服务发现、配置管理和服务治理平台。它支持多种语言的客户端，包括 Java、Python 等，能够帮助开发者轻松地实现服务的注册、发现和配置管理。

二、安装 Nacos

在开始之前，需要先安装并启动 Nacos 服务。可以通过以下几种方式安装 Nacos：

使用 Docker 安装：docker pull nacos/nacos-server docker run -d -p 8848:8848 --name nacos-server nacos/nacos-server Nacos 默认监听端口为 8848，访问 Nacos 控制台的 URL 为 http://localhost:8848/nacos。这里也可以直接使用源码运行,直接在本地idea上运行即可不需要依赖服务器环境或者虚拟机比较麻烦.下面有我自己根据微服务框架pig抽取出来的nacos源码,可以直接clone下来运行.

gitee

直接运行可执行文件：从 Nacos 官方网站下载最新版本的 Nacos，解压后运行 startup.sh 或 startup.bat 文件启动 Nacos。

三、Python 服务注册到 Nacos

要将 Python 服务注册到 Nacos，可以使用 nacos-sdk-python 客户端库。以下是具体步骤：

1. 安装 `nacos-sdk-python`

在你的 Python 项目中，运行以下命令安装 Nacos 的 Python SDK：

代码语言：bash复制

pip install nacos-sdk-python

2. 配置 Nacos 服务器信息

创建一个配置文件（如 nacos_config.py），用于存储 Nacos 服务器的地址和其他相关信息：

代码语言：python代码运行次数：0运行复制

# nacos_config.py
NACOS_SERVER_ADDRESSES = "http://localhost:8848"
NACOS_NAMESPACE = "public"
NACOS_GROUP = "DEFAULT_GROUP"
SERVICE_NAME = "your_service_name"

这里分别对应nacos的地址,naocs命名空间,nacos分组名,nacos的服务名

3. 实现服务注册逻辑

创建一个 Python 文件（如 nacos_client.py），用于实现服务注册逻辑：

代码语言：python代码运行次数：0运行复制

# nacos_client.py
from nacos import NacosClient
from nacos_config import NACOS_SERVER_ADDRESSES, NACOS_NAMESPACE, SERVICE_NAME

class NacosService:
    def __init__(self):
        self.client = NacosClient(NACOS_SERVER_ADDRESSES, namespace=NACOS_NAMESPACE)

    def register_service(self, ip, port, metadata=None):
        """注册服务到 Nacos"""
        self.client.add_naming_instance(
            service_name=SERVICE_NAME,
            ip=ip,
            port=port,
            metadata=metadata
        )
        print(f"Service {SERVICE_NAME} registered to Nacos at {ip}:{port}")

    def deregister_service(self, ip, port):
        """从 Nacos 注销服务"""
        self.client.remove_naming_instance(
            service_name=SERVICE_NAME,
            ip=ip,
            port=port
        )
        print(f"Service {SERVICE_NAME} deregistered from Nacos at {ip}:{port}")

4. 在 FastAPI 应用中集成服务注册

在 FastAPI 应用的入口文件（如 main.py）中，使用 @app.on_event 装饰器在应用启动和关闭时执行服务注册与注销逻辑：

代码语言：python代码运行次数：0运行复制

# main.py
from fastapi import FastAPI
from nacos_client import NacosService

app = FastAPI()

# Nacos 服务配置 这里指当前python服务的ip和端口
SERVICE_IP = "127.0.0.1"  # 你的服务 IP 地址
SERVICE_PORT = 8000       # 你的服务端口

# 初始化 Nacos 服务
nacos_service = NacosService()

@app.on_event("startup")
async def startup_event():
    """应用启动时注册服务"""
    nacos_service.register_service(ip=SERVICE_IP, port=SERVICE_PORT)
    print("Service registered to Nacos on startup")

@app.on_event("shutdown")
async def shutdown_event():
    """应用关闭时注销服务"""
    nacos_service.deregister_service(ip=SERVICE_IP, port=SERVICE_PORT)
    print("Service deregistered from Nacos on shutdown")

@app.get("/")
async def read_root():
    return {"message": "Hello, Nacos!"}

四、Python 服务发现

服务发现是指客户端从 Nacos 获取服务实例列表。以下是如何实现服务发现的代码示例：

1. 实现服务发现逻辑

在 nacos_client.py 中，添加服务发现逻辑：

代码语言：python代码运行次数：0运行复制

# nacos_client.py (继续)
class NacosService:
    # ...（之前的代码）

    def discover_service(self, service_name):
        """发现服务实例"""
        instances = self.client.list_naming_instance(service_name=service_name)
        print(f"Discovered instances for {service_name}: {instances}")
        return instances

2. 在客户端应用中使用服务发现

在客户端应用中，可以通过调用 discover_service 方法来获取服务实例列表：

代码语言：python代码运行次数：0运行复制

# client.py
from nacos_client import NacosService

nacos_service = NacosService()

# 发现服务
instances = nacos_service.discover_service(SERVICE_NAME)
for instance in instances:
    print(f"Service instance: {instance['ip']}:{instance['port']}")

最后我在这里贴上全部代码:

代码语言：python代码运行次数：0运行复制

# nacos_config.py
NACOS_SERVER_ADDRESSES = "http://localhost:8848"
NACOS_NAMESPACE = "public"
NACOS_GROUP = "DEFAULT_GROUP"
NACOS_DATA_ID = "demo-register"

代码语言：python代码运行次数：0运行复制

# nacos_client.py
from nacos import NacosClient
from nacos_config import NACOS_SERVER_ADDRESSES, NACOS_NAMESPACE, NACOS_GROUP, NACOS_DATA_ID
import yaml

class NacosService:
    def __init__(self, service_name):
        self.client = NacosClient(NACOS_SERVER_ADDRESSES, namespace=NACOS_NAMESPACE)
        self.service_name = service_name
        self.data_id = f"{service_name}.yaml"

    def register_service(self, ip, port, metadata=None):
        """注册服务到 Nacos"""
        self.client.add_naming_instance(
            service_name=self.service_name,
            ip=ip,
            port=port,
            metadata=metadata
        )
        print(f"Service {self.service_name} registered to Nacos at {ip}:{port}")

    def deregister_service(self, ip, port):
        """从 Nacos 注销服务"""
        self.client.remove_naming_instance(
            service_name=self.service_name,
            ip=ip,
            port=port
        )
        print(f"Service {self.service_name} deregistered from Nacos at {ip}:{port}")

    def get_config(self):
        """从 Nacos 获取配置"""
        config = self.client.get_config(self.data_id, NACOS_GROUP)
        if config:
            return yaml.safe_load(config)
        return None

代码语言：python代码运行次数：0运行复制

# main.py
# -*- coding: utf-8 -*-
import uvicorn
from controller import app
from nacos_client import NacosService

# Nacos 服务配置
SERVICE_IP = "0.0.0.0"
SERVICE_PORT = 5002
SERVICE_NAME = "python-demo"

# 初始化 Nacos 服务
nacos_service = NacosService(service_name=SERVICE_NAME)

@app.on_event("startup")
async def startup_event():
    """应用启动时注册服务"""
    nacos_service.register_service(ip=SERVICE_IP, port=SERVICE_PORT)
    print("Service registered to Nacos on startup")

    # 从 Nacos 获取配置
    config = nacos_service.get_config()
    if config:
        print(f"Config from Nacos: {config}")
        app.config = config
    else:
        print("No config found in Nacos")

@app.on_event("shutdown")
async def shutdown_event():
    """应用关闭时注销服务"""
    nacos_service.deregister_service(ip=SERVICE_IP, port=SERVICE_PORT)
    print("Service deregistered from Nacos on shutdown")

@app.get("/")
async def read_root():
    return {"message": "Hello, Nacos!"}

if __name__ == '__main__':
    uvicorn.run(app, host="0.0.0.0", port=5002)

五、注意事项

心跳机制：注册服务后，建议定期向 Nacos 发送心跳，以维持服务的在线状态。可以通过 nacos-sdk-python 提供的 send_heartbeat 方法实现。
配置管理：Nacos 还支持动态配置管理。可以通过 NacosClient.get_config 和 NacosClient.publish_config 方法读取和发布配置。
服务健康检查：可以通过 Nacos 的健康检查功能，确保服务实例的可用性。

通过以上步骤，你可以将 Python 服务成功注册到 Nacos 中，并实现服务的发现功能。这将帮助你在微服务架构中更好地管理服务实例，提高系统的可维护性和可扩展性。

启动nacos服务

这里我使用的是源码运行如下图:

启动python服务

可以发现日志打印已经显示注册成功了,我们再继续看看nacos的控制台

最后我们就完成了nacos注册python服务

发布者：admin，转转请注明出处：http://www.yc00.com/web/1747489407a4645924.html

如何使用 Nacos 对 Python 服务进行服务发现与注册