【Python】Python注册微服务到nacos_python nacos

Python注册微服务到Nacos

1.Nacos部署

• github 的nacos项目的发布页（Releases · alibaba/nacos · GitHub ），选择所要下载的nacos版本，在nacos下方的assets中选择安装包进行下载。

  

• 解压nacos安装包到指定目录。

  tar -zxvf nacos-server-2.0.3.tar.gz
  

• 初始化Nacos数据库。

  -- 需要先选择数据库
  -- 初始化sql脚本位于解压的 nacos/conf 目录下
  use nacosdb;
  source /euansu/apps/nacos/conf/nacos-mysql.sql;
  

  

• 修改nacos配置文件。

  # nacos/conf/application.properties
  # 默认的web路由
  server.servlet.contextPath=/nacos
  # web服务端口
  server.port=8848
  # 数据库配置
  db.url.0=jdbc:mysql://127.0.0.1:3306/nacos?characterEncoding=utf8&connectTimeout=1000&socketTimeout=3000&autoReconnect=true&useUnicode=true&useSSL=false&serverTimezone=UTC
  db.user.0=nacos
  db.password.0=nacos
  

• 启动Nacos服务，这里只是以单机模式启动Nacos。

  bin/startup.sh -m standalone
  

  

  启动完成之后，访问页面进行验证。

  

  如果没有在配置文件中修改密码，这里的用户名和密码是：nacos/nacos，直接登录即可。

  

2.Nacos验证

Nacos注册接口（注册一个实例服务到Naocs上）

• 请求地址：/nacos/v1/ns/instance

• 请求方式：POST

• 请求参数：

  名称	类型	是否必选	描述
  ip	字符串	是	服务实例IP
  port	int	是	服务实例port
  namespaceId	字符串	否	命名空间ID
  weight	double	否	权重
  enabled	boolean	否	是否上线
  healthy	boolean	否	是否健康
  metadata	字符串	否	扩展信息
  clusterName	字符串	否	集群名
  serviceName	字符串	是	服务名
  groupName	字符串	否	分组名
  ephemeral	boolean	否	是否临时实例

• 返回参数：

  返回代码	描述	语义
  400	Bad Request	客户端请求中的语法错误
  403	Forbidden	没有权限
  404	Not Found	无法找到资源
  500	Internal Server Error	服务器内部错误
  200	OK	正常

• 请求测试：

  

  请求接口后，在Nacos页面出现注册的服务，如下：

  

  查看详细信息

  

Nacos服务检测接口：

• 请求地址：/nacos/v1/ns/instance/beat

• 请求方式：PUT

• 请求参数：

  名称	类型	是否必选	描述
  serviceName	字符串	是	服务名
  ip	字符串	是	服务实例IP
  port	int	是	服务实例PORT
  namespaceId	字符串	否	命名空间ID
  groupName	字符串	否	分组名
  ephemeral	boolean	否	是否临时实例
  beat	JSON格式字符串	是	实例心跳内容

• 响应参数：

  响应代码	描述	语义
  400	Bad Request	客户端请求中的语法错误
  403	Forbidden	没有权限
  404	Not Found	无法找到资源
  500	Internal Server Error	服务器内部错误
  200	OK	正常

• 请求测试：

  

将Python服务注册到Nacos上至少需要两个接口，一个是实例注册接口，另一个则是心跳接口，只进行了注册，则会出现如下的情况：

时间稍长之后，注册的服务也会消失。

因此，需要按照心跳时间请求nacos服务检测接口，默认是5s。

3.Python注册Nacos服务

Python这里可以通过写一个脚本实现Nacos服务的注册，实现代码如下：

import time
import requests
​
# 心跳时间
HEARTBEATS_TIME = 5
# Nacos地址
NACOS_URL = 'http://xx.xx.xx.xx:8848/nacos'
# 服务注册路由
NACOR_REGISTER_URL = 'v1/ns/instance'
# 心跳检测路由
NACOS_HEARTBEATS_URL = 'v1/ns/instance/beat'
​
​
# Nacos服务注册
def service_register():
    """
    Nacos服务注册的接口有以下参数：
    ip          服务实例IP
    port        服务实例port
    namespaceId 命名空间ID
    weight      权重
    enabled     是否上线
    healthy     是否健康
    metadata    扩展信息
    clusterName 集群名
    serviceName 服务名
    groupName   分组名
    ephemeral   是否临时实例
    :return:
    """
    # 返回参数
    result = {
        "code": 200,
        "message": "success"
    }
    # 构造请求参数
    params = {
        'serviceName': 'python-service',
        'ip': 'xx.xx.xx.xx',
        'port': 8845
    }
    response = requests.post(
        "{NACOS_URL}/{NACOR_REGISTER_URL}".format(NACOS_URL=NACOS_URL, NACOR_REGISTER_URL=NACOR_REGISTER_URL),
        params=params)
    if response.status_code != 200:
        result["code"] = response.status_code
        result["message"] = "服务注册Nacos失败，失败原因：{error_message}".format(error_message=response.text)
        return result
    if response.text != 'ok':
        result["code"] = 500
        result["message"] = "服务注册Nacos失败，失败原因：{error_message}".format(error_message=response.text)
        return result
    return result
​
​
def service_beat():
    """
    服务心跳，默认是5s一次
    :return:
    nacos的服务心跳接口有以下参数
    serviceName 服务名
    ip          服务实例IP
    port        服务实例PORT
    namespaceId 命名空间ID
    groupName   分组名
    ephemeral   是否临时实例
    beat        实例心跳内容
    """
    # 构造请求参数
    params = {
        "serviceName": "python-service",
        'ip': 'xx.xx.xx.xx',
        'port': 8845
    }
    while True:
        response = requests.put(
            "{NACOS_URL}/{NACOS_HEARTBEATS_URL}".format(NACOS_URL=NACOS_URL, NACOS_HEARTBEATS_URL=NACOS_HEARTBEATS_URL),
            params=params)
        print("已注册服务，执行心跳服务，续期服务响应状态： {status_code}".format(status_code=response.status_code))
        time.sleep(HEARTBEATS_TIME)
​
def main():
    # 注册服务
    service_register()
    # 服务检测
    service_beat()
​
​
if __name__ == '__main__':
    main()

实现效果：

4.参考文档

nacos文档：Open API 指南