python flask服务如何注册到nacos

shigen坚持更新文章的博客写手，擅长Java、python、vue、shell等编程语言和各种应用程序、脚本的开发。记录成长，分享认知，留住感动。
个人IP：shigen

背景

shigen之前遇到了一个服务，需要结合nacos+ Spring security实现服务的负载均衡。其中最为重要的就是python服务自己注册到nacos上，趁着闲暇时间，来研究了一番。

在nacos官网中提到了【其它语言的sdk】：

nacos其它语言SDK

点击【python】就跳转到了GitHub。

官网内容

看着好像也没说的很清楚，我们自己来调用SDK实现一下。

代码实现

首先确保nacos服务启动了，shigen这里用的docker-compose的方式启动的：

nacos启动

对应的docker-compose文件如下：

version: '3'
services:nacos-standalone:image: nacos/nacos-server:v2.1.2-slimcontainer_name: nacos-serverports:- 8848:8848- 9848:9848- 9849:9849environment:- 'MODE=standalone'- 'TIME_ZONE=Asia/Shanghai'volumes:- ./conf:/home/nacos/conf- ./logs:/home/nacos/logs- ./data:/home/nacos/data

代码可以参考shigen的gitee-cloud-patform。

首先下载依赖：

pip install nacos-sdk-python

安装依赖

依赖安装完成，就可以继续写代码了，主要是注册上服务和发送心跳。

最开始，shigen尝试了官方的SDK，总是出现异常，最后看了一下官方的描述：

Supported Python version：

Python 2.7 Python 3.6 Python 3.7

Supported Nacos version

Nacos 0.8.0 ~ 1.3.2

好家伙，我忽视了nacos的版本，注意，我的nacos是这样的：nacos/nacos-server:v2.1.2-slim，直接2.0+，不适配。害得我折腾了一下午，最后和伙伴出去跑步了，心里一想，我真是钻牛角尖。服务的注册也不是我的首要任务，实在不行，用官方的API照样可以注册上去。

于是，回来尝试了一下：

import sys
# 用于异步处理心跳检测
import threading
import time
from typing import NoReturnimport requests
from flask import Flask, jsonify, requestapp = Flask(__name__)# 服务注册
def service_register(service_name: str, ip: str, port: int):url = f"http://127.0.0.1:8848/nacos/v1/ns/instance?serviceName={service_name}&ip={ip}&port={port}"res = requests.post(url)print(f"完成注册: {res.status_code}")# 心跳检测
def service_beat(service_name: str, ip: str, port: int) -> NoReturn:while True:url = f"http://127.0.0.1:8848/nacos/v1/ns/instance/beat?serviceName={service_name}&ip={ip}&port={port}"res = requests.put(url)print(f"心跳检测中... 响应状态码： {res.status_code}")time.sleep(5)@app.route('/get_server_info', methods=['GET'])
def get_server_info():server_ip = request.host.split(':')[0]server_port = request.host.split(':')[1]response = {'ip': server_ip, 'port': server_port}return jsonify(response)if __name__ == '__main__':if len(sys.argv) > 1:port = int(sys.argv[1])else:port = 10000service_name = 'python-server'ip = '127.0.0.1'service_register(service_name, ip, port)# 5 秒后执行心跳检测threading.Timer(5, service_beat(service_name, ip, port)).start()app.run()