Python发布微服务到注册中心Nacos指南_Python

1.准备工作

1.1.安装python环境

https://www.python.org/,在官网下载python安装包以后，进行安装，配置好对应的环境变量（ps使用安装包安装，界面有添加到python到环境变量选项）。

1.2.安装flask框架

该框架（函数库）是python的一款mvc框架，相信做web端开发过来的，都比较容易理解，尤其java，例如springmvc框架。

这个主要是用来开发http接口的。

pip install flask

1.3.安装requests函数库

这个是用来发起http请求的，因为我们如果将服务发布到注册中心的话，需要我们向注册中心发起http请求。

pip install requests

1.4.安装注册中心nacos

安装略

nacos服务端访问地址：http://:8848/nacos。具体nacos学习参考如下两个
nacos github地址：https://github.com/alibaba/nacos
nacos 官方中文学习网：https://nacos.io/zh-cn/docs/what-is-nacos.html

2.python发布http/rest接口

创建demo.py

2.1.发布http+get

创建一个py文件，例如demo.py

from flask import flask,jsonify,request
'''python发布restful服务实例
@author:hutao
@date: 2022-04-11
'''
#初始化。官方解释：http://flask.pocoo.org/docs/0.12/api/#flask.flask
server = flask(__name__)
#处理乱码
server.config['json_as_ascii']=false
'''使用rest进行get请求，通过请求拼接参数进行传参，其中name是传参
请求实例：http://127.0.0.1:8085/simulation/analysis?name=胡涛
请求参数：请求地址中的name=胡涛
响应参数：
{
    "code": "200",
    "data": {
        "age": 25,
        "job": "python",
        "name": "胡涛"
    },
    "msg": "succes"
}
'''
@server.route('/simulation/analysis',methods=['get'])
def demo_rest_get_request():
    # 处理业务逻辑
    name = request.args['name']
    result = {"code":"200","msg":"succes","data":{"name":name,"age":25,"job":"python"}}
    return jsonify(result)
if __name__ == "__main__":
    server.run(port=8085,debug=true)

启动该demo.py文件，然后通过postman工具调用这个接口，效果如下所示。

2.2.发布http+post+json

'''
使用rest进行post请求，通过请求提 json传参，其中name是传参
请求地址：http://127.0.0.1:8085/simulation/analysis
请求参数：
{
    "name":"胡涛",
    "job":"java"
}
响应参数：
{
    "code": "200",
    "data": {
        "age": 25,
        "job": "java",
        "name": "胡涛"
    },
    "msg": "succes"
}
'''
@server.route('/simulation/analysis',methods=['post'])
def demo_rest_post_request():
    # 处理业务逻辑
    name = request.json.get('name')
    job = request.json.get('job')
    result = {"code":"200","msg":"succes","data":{"name":name,"age":25,"job":job}}
    return jsonify(result)

2.3.发布restful

'''使用restful进行get请求，通过请求地址进行传参，其中胡涛是传参
请求地址：http://127.0.0.1:8085/simulation/analysis/胡涛
请求参数：将地址中的胡涛映射到属性字段name上
响应参数：
{
    "code": "200",
    "data": {
        "age": 26,
        "job": "python",
        "name": "胡涛"
    },
    "msg": "succes"
}
'''
@server.route('/simulation/analysis/<name>',methods=['get'])
def demo_restful_request(name):
    result = {"code":"200","msg":"succes","data":{"name":name,"age":25,"job":"python"}}
return jsonify(result)

3.java通过rest请求接口

//调用python服务
string url = "http://127.0.0.1:8085/simulation/analysis?name=hutao";
responseentity<restdata> forentity = resttemplate.getforentity(url, restdata.class);
restdata<pythondata> data = forentity.getbody();

4.python服务注册

4.1.nacos注册接口

通过postman对nacos服务发起注册请求即可。

请求地址：http://127.0.0.1:8848/nacos/v1/ns/instance
请求方式：post

请求参数：

备注：请求参数并不是json格式，不同版本，可能会存在差异，本人使用的是1.4.3版本，请求参数并不能通过json格式提交。而是在请求地址后面拼接参数。

http://49.232.70.33:8848/nacos/v1/ns/instance?servicename=algorithm-service&ip=127.0.0.1&port=8085

不过这个不是重点，具体什么请求参数，什么请求方式，这个由于nacos开源，可以自己去定义的。

该接口返回ok以后，即注册成功，此时登陆注册中心，即可看到我们注册的服务。

4.2.nacos心跳接口

上述注册的接口，会在一段时间内和注册中心断开，因此注册中心没有收到客户端（注册的服务）的心跳，因此判断该服务出问题，该版本大概是15秒左右，就会变成如下这样。

过一段时间后，任然没有心跳，则会在服务列表剔除。

因此为了保证我们的服务注册上以后，保持健康状态，即让注册中心知道我们的服务没有问题，网络可达，因此需要定期向注册中心发起心跳检测的请求。

请求地址：http://127.0.0.1:8848/nacos/v1/ns/instance/beat
请求方式：put
请求参数：同注册

http://127.0.0.1:8848/nacos/v1/ns/instance/beat?servicename=algorithm-service&ip=127.0.0.1&port=8085

4.3.python实现思路

我们已经知道了通过上述两个接口，即可实现注册和心跳检测，因此，我们只需要将我们的python服务发布http接口的时候，给注册中心发起一个请求进行注册即可，接着在通过一个异步任务定时的向注册中心发起心跳检测即可。

当然，目前也有一些开源的软件实现了，不过笔者并不推荐，因为从技术上来说，这个真的很简单，自己实现即可，就是python给注册中心发起请求而已，并没有什么太复杂的功能。

例如目前开源了nacos-sdk-python，不过由于不同版本的nacos可能存在版本差异，导致到时候调试或者开发存在其他问题，因此不推荐使用。因为nacos开源，并且我们安装的nacos可能会是我们定制化后的。因此推荐你直接调用你所部署的nacos的服务接口

4.3.1.调用nacos注册接口

#nacos服务注册
def service_register():
    url = "http://49.232.70.33:8848/nacos/v1/ns/instance?servicename=algorithm-service&ip=127.0.0.1&port=8085"
    res = requests.post(url)
    print("向nacos注册中心，发起服务注册请求，注册响应状态： {}".format(res.status_code))

4.3.2.调用nacos心跳接口

#服务检测（每5秒心跳一次）
def service_beat():
    while true:
        url = "http://49.232.70.33:8848/nacos/v1/ns/instance/beat?servicename=algorithm-service&ip=127.0.0.1&port=8085"
        res = requests.put(url)
        print("已注册服务，执行心跳服务，续期服务响应状态： {}".format(res.status_code))
        time.sleep(5)

4.4.完整代码示例

from flask import flask,jsonify,request
import requests
import threading
import time
'''python发布restful服务实例
@author:hutao
@date: 2022-04-11
'''
# flask初始化参数尽量使用你的包名，这个初始化方式是官方推荐的，官方解释：http://flask.pocoo.org/docs/0.12/api/#flask.flask
server = flask(__name__)
#处理乱码
server.config['json_as_ascii']=false

'''使用restful进行get请求，通过请求地址进行传参，其中胡涛是传参
请求地址：http://127.0.0.1:8085/simulation/analysis/胡涛
请求参数：将地址中的胡涛映射到属性字段name上
响应参数：
{
    "code": "200",
    "data": {
        "age": 25,
        "job": "python",
        "name": "胡涛"
    },
    "msg": "succes"
}
'''
@server.route('/simulation/analysis/<name>',methods=['get'])
def demo_restful_request(name):
    # 处理业务逻辑
    #name = request.args['name']
    result = {"code":"200","msg":"succes","data":{"name":name,"age":25,"job":"python"}}
    return jsonify(result)

'''使用rest进行get请求，通过请求拼接参数进行传参，其中name是传参
请求实例：http://127.0.0.1:8085/simulation/analysis?name=胡涛
请求参数：请求地址中的name=胡涛
响应参数：
{
    "code": "200",
    "data": {
        "age": 25,
        "job": "python",
        "name": "胡涛"
    },
    "msg": "succes"
}
'''
@server.route('/simulation/analysis',methods=['get'])
def demo_rest_get_request():
    # 处理业务逻辑
    name = request.args['name']
    result = {"code":"200","msg":"succes","data":{"name":name,"age":25,"job":"python"}}
    return jsonify(result)


'''使用rest进行post请求，通过请求提 json传参，其中name是传参
请求地址：http://127.0.0.1:8085/simulation/analysis
请求参数：
{
    "name":"胡涛",
    "job":"java"
}
响应参数：
{
    "code": "200",
    "data": {
        "age": 25,
        "job": "java",
        "name": "胡涛"
    },
    "msg": "succes"
}

'''
@server.route('/simulation/analysis',methods=['post'])
def demo_rest_post_request():
    # 处理业务逻辑
    name = request.json.get('name')
    job = request.json.get('job')
    result = {"code":"200","msg":"succes","data":{"name":name,"age":25,"job":job}}
    return jsonify(result)


#nacos注册中心信息
'''将服务注册到注册中心
注册说明：将http://127.0.0.1:8085/**这个服务上的所有服务注册到注册中心，并且起名叫做algorithm-service
其他微服务进行访问时，访问http://algorithm-service/**即可，即其他服务，使用algorithm-service去注册中心，寻找真实的ip地址
例如原本访问 post访问：http://127.0.0.1:8085/simulation/analysis 此时变成 http://algorithm-service/simulation/analysis
'''
#nacos服务
def service_register():
    url = "http://49.232.70.33:8848/nacos/v1/ns/instance?servicename=algorithm-service&ip=127.0.0.1&port=8085"
    res = requests.post(url)

    print("向nacos注册中心，发起服务注册请求，注册响应状态： {}".format(res.status_code))
#服务检测
def service_beat():
    while true:
        url = "http://49.232.70.33:8848/nacos/v1/ns/instance/beat?servicename=algorithm-service&ip=127.0.0.1&port=8085"
        res = requests.put(url)
        print("已注册服务，执行心跳服务，续期服务响应状态： {}".format(res.status_code))
        time.sleep(5)

#发布http服务，并且注册到nocos
if __name__ == "__main__":
    service_register()
    #5秒以后，异步执行service_beat()方法
    threading.timer(5,service_beat).start()
    server.run(port=8085,debug=true)