Python通过HTTP方式的OpenAPI注册Nacos的实例

目录1. 背景2. 注册Nacos的实例1. 背景虽然Python提供了Nacos的SDK，但是该SDK已经不再维护更新了所以只能通过HTTP方式的OpenAPI进行Nacos的相关操作，具体的API参考Nacos Open API 指南2. 注册Nacos的实例功能是在Nacos服务(Ip = 192.168.8.246，Port = 8848)中，注册提供真实后端服务(Ip = 192.168

Bulut0907

2246人浏览 · 2022-06-17 17:37:12

Bulut0907 · 2022-06-17 17:37:12 发布

1. 背景

虽然Python提供了Nacos的SDK，但是该SDK已经不再维护更新了

所以只能通过HTTP方式的OpenAPI进行Nacos的相关操作，具体的API参考Nacos Open API 指南

2. 注册Nacos的实例

功能是在Nacos服务(Ip = 192.168.8.246，Port = 8848)中，注册提供真实后端服务(Ip = 192.168.8.115，Port = 26400)的服务发现，注册的serviceName = data-quality-system

下面的程序共分为3步：

登录Nacos，获取accessToken
注册Nacos实例
通过子线程，不停的向Nacos发送实例的心跳

实操代码如下：

import requests
import json
import time
from multiprocessing import Pool

def asyncInstanceBeat(accessToken):
    # 持续发送心跳
    beatDict = {"cluster":"DEFAULT", "ip":"192.168.8.115", "metadata":{}, "port":26400,
                "scheduled":True, "serviceName":"data-quality-system", "weight":1}
    instanceBeatData = {'serviceName':'data-quality-system', 'groupName':'DEFAULT_GROUP', 'ephemeral':True,
                        'accessToken':accessToken, 'beat':json.dumps(beatDict)}
    instanceBeatUrl = "http://192.168.8.246:8848/nacos/v1/ns/instance/beat"

    while (True):
        instanceBeatResponse = requests.put(instanceBeatUrl, instanceBeatData)
        print("instanceBeatResponse: " + instanceBeatResponse.text)
        time.sleep(5)

def registerServerToNacos():
    # 登录nacos获取accessToken
    loginUrl = "http://192.168.8.246:8848/nacos/v1/auth/users/login"
    loginData = {'username':'hnmqet', 'password':'123456', 'namespace':'public'}
    loginResponse = requests.post(loginUrl, data= loginData)
    accessToken = dict(json.loads(loginResponse.text)).get("accessToken")
    print("accessToken: " + accessToken)

    # 注册实例
    registerInstanceUrl = "http://192.168.8.246:8848/nacos/v1/ns/instance"
    registerInstanceData={'serviceName':'data-quality-system', 'ip':'192.168.8.115', 'port':'26400',
                          'clusterName':'DEFAULT', 'weight':1, 'enable':True, 'healthy':True,
                          'ephemeral':True, 'groupName':'DEFAULT_GROUP',
                          'accessToken':accessToken}
    registerInstanceResponse = requests.post(registerInstanceUrl, registerInstanceData)
    print("registerInstance:" + registerInstanceResponse.text)

    threadPool.apply_async(asyncInstanceBeat, args=(accessToken,))

if __name__ == '__main__':

    threadPool = Pool(1)
    registerServerToNacos(threadPool)

    threadPool.close()
    threadPool.join()

运行程序，打印的日志如下：

accessToken: eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJobm1xZXQiLCJleHAiOjE2NTI3MDQxOTB9.v_3hzLzxHTPcvvdysU5BGbjRHzn-vGpMhagOKv-36Ek
registerInstance:ok
instanceBeatResponse: {"clientBeatInterval":5000,"code":10200,"lightBeatEnabled":true}
instanceBeatResponse: {"clientBeatInterval":5000,"code":10200,"lightBeatEnabled":true}
instanceBeatResponse: {"clientBeatInterval":5000,"code":10200,"lightBeatEnabled":true}
......省略部分......