nacos_client.py
2.5 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
"""
Nacos 服务注册客户端
使用 nacos-sdk-python v2，内置自动心跳
"""
import logging
import socket
import nacos
from .settings import get_settings
logger = logging.getLogger(__name__)
_client: nacos.NacosClient | None = None
def _get_local_ip() -> str:
"""获取本机内网 IP"""
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.connect(("8.8.8.8", 80))
ip = s.getsockname()[0]
s.close()
return ip
except Exception:
return "127.0.0.1"
def register_service() -> None:
"""注册服务到 Nacos（v2 内置自动心跳）"""
global _client
settings = get_settings()
if not settings.nacos_enabled:
logger.info("Nacos 注册已禁用，跳过")
return
server_addr = settings.nacos_server_addr
namespace = settings.nacos_namespace or ""
_client = nacos.NacosClient(
server_addr,
namespace=namespace,
)
ip = _get_local_ip()
port = settings.nacos_service_port
service_name = settings.nacos_service_name
group = settings.nacos_group
metadata = {
"group": settings.nacos_metadata_group,
"region": settings.nacos_metadata_region,
"version": settings.nacos_metadata_version,
"preserved.register.source": "python",
}
try:
_client.add_naming_instance(
service_name,
ip,
port,
group_name=group,
metadata=metadata,
enable=True,
healthy=True,
ephemeral=True,
heartbeat_interval=5,
)
logger.info(
f"Nacos 注册成功: {service_name} -> {ip}:{port} "
f"(namespace={namespace}, group={group}, 心跳间隔=5s)"
)
except Exception as e:
logger.error(f"Nacos 注册失败: {e}", exc_info=True)
def deregister_service() -> None:
"""从 Nacos 注销服务"""
global _client
settings = get_settings()
if not settings.nacos_enabled or _client is None:
return
ip = _get_local_ip()
port = settings.nacos_service_port
service_name = settings.nacos_service_name
group = settings.nacos_group
try:
_client.remove_naming_instance(
service_name,
ip,
port,
group_name=group,
)
logger.info(f"Nacos 注销成功: {service_name} -> {ip}:{port}")
except Exception as e:
logger.error(f"Nacos 注销失败: {e}", exc_info=True)
finally:
_client = None