Consul测试代码

import time
from consul import Consul, Checkclass ConsulServiceManager:def __init__(self, host, port=8500, schema="http"):self._host = hostself._port = portself._schema = schemaself._local_ip = "127.0.0.1"  # 通过环境变量获取self.consul = Consul(host=self._host, port=self._port, scheme=self._schema)def register_service(self, service_name, service_id, service_address, service_port):self.consul.agent.service.register(name=service_name,service_id=service_id,address=service_address,port=service_port,check=Check.http(f'https://{service_address}:{service_port}/health',interval='10s',timeout='5s',),)print(f'Service {service_name} registered with ID {service_id}.')def deregister_service(self, service_id):self.consul.agent.service.deregister(service_id)print(f'Service with ID {service_id} deregistered.')def discover_service(self, service_name):index, services = self.consul.catalog.services()print("Available services:")for service in services:print(f'- {service}')# 获取具体服务的信息index, service_info = self.consul.catalog.service(service_name)if service_info:print(f"Service '{service_name}' found with info: {service_info}")else:print(f"Service '{service_name}' not found.")if __name__ == '__main__':# 创建 ConsulServiceManager 实例consul_manager = ConsulServiceManager(host='192.168.20.50')# 服务信息_service_name = 'my_service'_service_id = 'my_service_id'_service_address = 'www.xxx.com'  # 使用 Consul 的主机名_service_port = 8080# 注册服务
    consul_manager.register_service(_service_name, _service_id, _service_address, _service_port)# 等待一段时间，以便服务注册time.sleep(5)# 发现服务
    consul_manager.discover_service(_service_name)# 注销服务time.sleep(5)consul_manager.deregister_service(_service_id)