文档建议反馈控制台
首页
学习
活动
专区
圈层
工具
MCP广场
文章/答案/技术大牛
发布
社区首页 >专栏 >如何使用 Nacos 对 Python 服务进行服务发现与注册

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

原创
作者头像
全干程序员demo
发布于 2025-04-28 10:12:07
发布于 2025-04-28 10:12:07
1.2K0
举报
文章被收录于专栏：技术热文技术热文

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

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

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

一、Nacos 简介

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

二、安装 Nacos

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

  1. 使用 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 https://gitee.com/Demotutu/demo-register

  1. 直接运行可执行文件： 从 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
复制
# 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
复制
# 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
复制
# 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
复制
# 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
复制
# 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
复制
# nacos_config.py
NACOS_SERVER_ADDRESSES = "http://localhost:8848"
NACOS_NAMESPACE = "public"
NACOS_GROUP = "DEFAULT_GROUP"
NACOS_DATA_ID = "demo-register"
代码语言：python
复制
# 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
复制
# 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)

五、注意事项

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

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

启动nacos服务

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

启动python服务

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

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

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

python
nacos

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

python
nacos
评论
登录后参与评论
0 条评论
热度
最新
登录 后参与评论
推荐阅读
目录
  • 如何使用 Nacos 对 Python 服务进行服务发现与注册
    • 一、Nacos 简介
    • 二、安装 Nacos
    • 三、Python 服务注册到 Nacos
      • 1. 安装 nacos-sdk-python
      • 2. 配置 Nacos 服务器信息
      • 3. 实现服务注册逻辑
      • 4. 在 FastAPI 应用中集成服务注册
    • 四、Python 服务发现
      • 1. 实现服务发现逻辑
      • 2. 在客户端应用中使用服务发现
    • 五、注意事项
    • 启动nacos服务
    • 启动python服务
领券

腾讯云开发者

扫码关注腾讯云开发者

扫码关注腾讯云开发者

领取腾讯云代金券

热门产品

热门推荐

更多推荐

Copyright © 2013 - 2026 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有 

深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 粤公网安备44030502008569号

腾讯云计算（北京）有限责任公司 京ICP证150476号 |  京ICP备11018762号

问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档

Copyright © 2013 - 2026 Tencent Cloud.

All Rights Reserved. 腾讯云 版权所有

登录 后参与评论