01.nacos-sdk-python测试

• nacos-sdk-python官方地址 (opens new window)

• nacos-sdk-python github地址 (opens new window)

• 安装 nacos-sdk-python

pip install nacos-sdk-python

1.1 nacos_cli/nacos_agent.py

• 对官方 NacosClient 简单封装

import random
from retrying import retry
from nacos import NacosClient
from nacos_cli.utils import StartNacosBeat


class RNacosClient(NacosClient):

    # 重写特殊的构造方法，以便以后添加新特性（这里暂时没有用）
    def __init__(self, server_addresses, endpoint=None, namespace=None, ak=None, sk=None, username=None, password=None):
        NacosClient.__init__(self, server_addresses, endpoint=endpoint,
                             namespace=namespace, ak=ak, sk=sk, username=username, password=password)

    # 注册服务到 nacos 中，并使用守护线程启动心跳协议
    def rewrite_add_naming_instance(self, service_name, ip, port, cluster_name=None, weight=1.0,
                                    metadata=None, enable=True, healthy=True, ephemeral=True, group_name='DEFAULT_GROUP'):
        self.add_naming_instance(service_name, ip, port, cluster_name=cluster_name,weight=weight,
            metadata=metadata, enable=enable, healthy=healthy, ephemeral=ephemeral, group_name=group_name)
        StartNacosBeat({
            'service_name': service_name, 'ip': ip, 'port': port, 'cluster_name': cluster_name,
            'weight': weight, 'metadata': metadata, 'ephemeral': ephemeral, 'group_name': group_name},
            self).call_nacos_beat()

    # 使用简单随机法，获取一个可用的 ip,port
    @retry(stop_max_attempt_number=3, wait_incrementing_increment=8000)
    def select_one_instance(self, service_name, clusters=None, namespace_id=None, group_name=None, healthy_only=False):
        list_instance = self.list_naming_instance(service_name, clusters, namespace_id, group_name, healthy_only)
        ramdom_server = random.choice(list_instance['hosts'])
        return ramdom_server['ip'], ramdom_server['port']

1.2 nacos_cli/utils.py

• 由于官方sdk注册没有与nacos发心跳，这里自己启动心跳服务

import threading
from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime

# 启动nacos心跳服务
class StartNacosBeat(object):
    instance = None
    event_lock = threading.Event()

    def __new__(cls, *args, **kwargs):
        if StartNacosBeat.instance:
            return StartNacosBeat.instance
        else:
            StartNacosBeat.instance = object.__new__(cls)
            return StartNacosBeat.instance

    def __init__(self, data, nacos_client, seconds=5):
        self.data = data
        self.nacos_client = nacos_client
        self.seconds = seconds

    def call_nacos_beat(self):
        if not self.event_lock.is_set():
            t = threading.Thread(target=self.start_nacos_beat, args=())
            t.setDaemon(True)
            t.start()
        self.event_lock.set()

    def start_nacos_beat(self):
        sched = BlockingScheduler()
        sched.add_job(self.nacos_beat_job, 'interval', seconds=self.seconds)
        sched.start()

    def nacos_beat_job(self):
        print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "-------------> 发送心跳给nacos服务！！！")
        self.nacos_client.send_heartbeat(**self.data)

1.3 nacos_cli/example.py

• 使用封装的 RNacosClient ，其他功能直接继承的官方

import time
from nacos_cli.nacos_agent import RNacosClient


if __name__ == '__main__':
    # nacos官方：https://github.com/nacos-group/nacos-sdk-python
    # 1、nacos服务信息
    SERVER_ADDRESSES = "127.0.0.1:8848"  # nacos服务地址
    NAMESPACE = "e813bfe8-198c-425a-b503-e567566e8565"  # nacos中创建的空间id
    # 2、要注册的服务信息
    service_name = "test.service1"
    ip = "127.0.0.1"
    port = 50001
    cluster_name = "testCluster2"

    # 1）实例化 NacosClient
    client = RNacosClient(SERVER_ADDRESSES, namespace=NAMESPACE)
    # 2）注册服务 并且自动开启 心跳服务
    client.rewrite_add_naming_instance(service_name=service_name, ip=ip, port=port, cluster_name=cluster_name)
    # 3）获取注册服务 ip 列表
    server_ips = client.select_one_instance(service_name=service_name, healthy_only=True)
    print("从nacos获取可用ip:port---->", server_ips)
    time.sleep(400)

• 运行结果

xiaonaiqiang1@ZBMac-C02CW08SM nacos_cli %  python3 example.py 
从nacos获取可用ip:port----> ('127.0.0.1', 50001)
2021-12-14 14:22:16 -------------> 发送心跳给nacos服务！！！
2021-12-14 14:22:21 -------------> 发送心跳给nacos服务！！！

1.4 在nacos页面查看注册情况