怎样实现服务发现脚本

wen 实用脚本 2026-06-27 232

本文目录导读：

怎样实现服务发现脚本

基于DNS的服务发现
基于Consul的服务发现
基于Eureka的服务发现
基于etcd的服务发现
Shell脚本实现（简单版）
使用建议

我来详细介绍如何实现服务发现脚本,包括几种常见的方法和实现方案。

基于DNS的服务发现

Python实现示例

#!/usr/bin/env python3
import socket
import dns.resolver
import json
from typing import Dict, List, Optional
class DNSDiscovery:
    """基于DNS的服务发现"""
    def __init__(self, domain: str):
        self.domain = domain
        self.resolver = dns.resolver.Resolver()
    def discover_services(self, service_type: str = "_http") -> List[str]:
        """发现服务实例"""
        try:
            # SRV记录查询
            srv_record = f"{service_type}._tcp.{self.domain}"
            answers = self.resolver.resolve(srv_record, 'SRV')
            services = []
            for rdata in answers:
                # TXT记录查询（获取元数据）
                txt_records = self.get_txt_records(rdata.target.to_text())
                services.append({
                    'host': str(rdata.target).rstrip('.'),
                    'port': rdata.port,
                    'priority': rdata.priority,
                    'weight': rdata.weight,
                    'metadata': txt_records
                })
            return services
        except Exception as e:
            print(f"服务发现失败: {e}")
            return []
    def get_txt_records(self, hostname: str) -> Dict:
        """获取TXT记录"""
        try:
            answers = self.resolver.resolve(hostname, 'TXT')
            metadata = {}
            for rdata in answers:
                for txt_string in rdata.strings:
                    if '=' in txt_string.decode():
                        key, value = txt_string.decode().split('=', 1)
                        metadata[key] = value
            return metadata
        except:
            return {}
# 使用示例
discoverer = DNSDiscovery("example.com")
services = discoverer.discover_services("_http")
print(json.dumps(services, indent=2))

基于Consul的服务发现

#!/usr/bin/env python3
import requests
import json
import os
from typing import Dict, List, Optional
class ConsulDiscovery:
    """基于Consul的服务发现"""
    def __init__(self, consul_host: str = "localhost", consul_port: int = 8500):
        self.consul_url = f"http://{consul_host}:{consul_port}"
    def register_service(self, 
                        service_name: str,
                        service_id: str,
                        address: str,
                        port: int,
                        tags: List[str] = None,
                        check: Dict = None) -> bool:
        """注册服务"""
        service = {
            "ID": service_id,
            "Name": service_name,
            "Address": address,
            "Port": port,
            "Tags": tags or [],
            "Check": check or {
                "TCP": f"{address}:{port}",
                "Interval": "10s",
                "Timeout": "5s"
            }
        }
        try:
            response = requests.put(
                f"{self.consul_url}/v1/agent/service/register",
                json=service
            )
            return response.status_code == 200
        except Exception as e:
            print(f"服务注册失败: {e}")
            return False
    def discover_service(self, service_name: str, healthy_only: bool = True) -> List[Dict]:
        """发现服务实例"""
        try:
            url = f"{self.consul_url}/v1/health/service/{service_name}"
            if healthy_only:
                url += "?passing"
            response = requests.get(url)
            if response.status_code != 200:
                return []
            services = []
            data = response.json()
            for service_info in data:
                service = service_info.get('Service', {})
                services.append({
                    'id': service.get('ID'),
                    'name': service.get('Service'),
                    'address': service.get('Address'),
                    'port': service.get('Port'),
                    'tags': service.get('Tags', []),
                    'node': service_info.get('Node', {}).get('Node'),
                    'health_status': self._get_health_status(service_info)
                })
            return services
        except Exception as e:
            print(f"服务发现失败: {e}")
            return []
    def _get_health_status(self, service_info: Dict) -> str:
        """获取健康状态"""
        checks = service_info.get('Checks', [])
        for check in checks:
            if check.get('CheckID') != 'serfHealth':
                if check.get('Status') != 'passing':
                    return check.get('Status', 'unknown')
        return 'passing'
    def deregister_service(self, service_id: str) -> bool:
        """注销服务"""
        try:
            response = requests.put(
                f"{self.consul_url}/v1/agent/service/deregister/{service_id}"
            )
            return response.status_code == 200
        except:
            return False
# 使用示例
consul = ConsulDiscovery()
consul.register_service(
    "web-app",
    "web-app-1",
    "192.168.1.100",
    8080,
    tags=["production", "v1.0"]
)
services = consul.discover_service("web-app")
print(json.dumps(services, indent=2))

基于Eureka的服务发现

#!/usr/bin/env python3
import requests
import json
from typing import Dict, List, Optional
class EurekaDiscovery:
    """基于Eureka的服务发现"""
    def __init__(self, eureka_url: str = "http://localhost:8761/eureka"):
        self.eureka_url = eureka_url.rstrip('/')
        self.headers = {
            'Content-Type': 'application/json',
            'Accept': 'application/json'
        }
    def register_service(self,
                        app_name: str,
                        instance_id: str,
                        hostname: str,
                        ip_addr: str,
                        port: int,
                        vip_address: str = None,
                        secure_vip_address: str = None) -> bool:
        """注册服务实例"""
        instance = {
            "instance": {
                "instanceId": instance_id,
                "hostName": hostname,
                "app": app_name.upper(),
                "ipAddr": ip_addr,
                "status": "UP",
                "port": {
                    "$": port,
                    "@enabled": "true"
                },
                "vipAddress": vip_address or hostname,
                "secureVipAddress": secure_vip_address or hostname,
                "dataCenterInfo": {
                    "@class": "com.netflix.appinfo.InstanceInfo$DefaultDataCenterInfo",
                    "name": "MyOwn"
                }
            }
        }
        try:
            url = f"{self.eureka_url}/apps/{app_name.upper()}"
            response = requests.post(
                url,
                json=instance,
                headers=self.headers
            )
            return response.status_code == 204
        except Exception as e:
            print(f"服务注册失败: {e}")
            return False
    def discover_service(self, app_name: str) -> List[Dict]:
        """发现服务实例"""
        try:
            url = f"{self.eureka_url}/apps/{app_name.upper()}"
            response = requests.get(url, headers=self.headers)
            if response.status_code != 200:
                return []
            data = response.json()
            apps = data.get('application', {})
            instances = apps.get('instance', [])
            if not isinstance(instances, list):
                instances = [instances]
            services = []
            for instance in instances:
                services.append({
                    'instance_id': instance.get('instanceId'),
                    'hostname': instance.get('hostName'),
                    'ip': instance.get('ipAddr'),
                    'port': instance.get('port', {}).get('$'),
                    'status': instance.get('status'),
                    'app': instance.get('app')
                })
            return services
        except Exception as e:
            print(f"服务发现失败: {e}")
            return []
    def send_heartbeat(self, app_name: str, instance_id: str) -> bool:
        """发送心跳"""
        try:
            url = f"{self.eureka_url}/apps/{app_name.upper()}/{instance_id}"
            response = requests.put(url, headers=self.headers)
            return response.status_code == 200
        except:
            return False
# 使用示例
eureka = EurekaDiscovery()
eureka.register_service(
    "user-service",
    "user-service-1",
    "node1.example.com",
    "192.168.1.101",
    8081
)
services = eureka.discover_service("user-service")
print(json.dumps(services, indent=2))

基于etcd的服务发现

#!/usr/bin/env python3
import etcd3
import json
import time
from typing import Dict, List, Optional
class EtcdDiscovery:
    """基于etcd的服务发现"""
    def __init__(self, host: str = "localhost", port: int = 2379):
        self.client = etcd3.client(host=host, port=port)
        self.base_path = "/services"
    def register_service(self,
                        name: str,
                        instance_id: str,
                        address: str,
                        port: int,
                        metadata: Dict = None,
                        ttl: int = 60) -> bool:
        """注册服务实例"""
        service_path = f"{self.base_path}/{name}/{instance_id}"
        service_info = {
            "name": name,
            "id": instance_id,
            "address": address,
            "port": port,
            "metadata": metadata or {},
            "timestamp": time.time()
        }
        try:
            # 设置带TTL的key
            self.client.put(
                service_path,
                json.dumps(service_info).encode(),
                lease=self.client.lease(ttl)
            )
            return True
        except Exception as e:
            print(f"服务注册失败: {e}")
            return False
    def discover_service(self, name: str) -> List[Dict]:
        """发现服务实例"""
        service_path = f"{self.base_path}/{name}/"
        try:
            instances = []
            for item in self.client.get_prefix(service_path):
                key, value = item
                service_info = json.loads(value.decode())
                instances.append(service_info)
            return instances
        except Exception as e:
            print(f"服务发现失败: {e}")
            return []
    def watch_service_changes(self, name: str, callback):
        """监听服务变化"""
        service_path = f"{self.base_path}/{name}/"
        events_iterator, cancel = self.client.watch_prefix(service_path)
        for event in events_iterator:
            if event.key:
                service_info = json.loads(event.value.decode())
                callback(event.key.decode(), event.type, service_info)
    def deregister_service(self, name: str, instance_id: str) -> bool:
        """注销服务实例"""
        service_path = f"{self.base_path}/{name}/{instance_id}"
        try:
            self.client.delete(service_path)
            return True
        except:
            return False
    def get_all_services(self) -> Dict[str, List[Dict]]:
        """获取所有服务"""
        try:
            services = {}
            for item in self.client.get_prefix(self.base_path):
                key, value = item
                key_str = key.decode()
                service_name = key_str.split('/')[2]
                if service_name not in services:
                    services[service_name] = []
                services[service_name].append(json.loads(value.decode()))
            return services
        except Exception as e:
            print(f"获取服务列表失败: {e}")
            return {}
# 使用示例
etcd = EtcdDiscovery()
etcd.register_service(
    "api-gateway",
    "gateway-1",
    "192.168.1.200",
    3000,
    metadata={"version": "1.0", "environment": "production"}
)
services = etcd.discover_service("api-gateway")
print(json.dumps(services, indent=2))

Shell脚本实现（简单版）

#!/bin/bash
# service_discovery.sh - 简单的服务发现脚本
# 服务注册
register_service() {
    local service_name=$1
    local service_host=$2
    local service_port=$3
    local service_id="${service_name}-${service_host}:${service_port}"
    # 简单文件存储
    echo "${service_host}:${service_port}" > "/tmp/services/${service_name}/${service_id}"
    echo "服务 ${service_id} 注册成功"
}
# 服务发现
discover_service() {
    local service_name=$1
    echo "发现服务: ${service_name}"
    if [ -d "/tmp/services/${service_name}" ]; then
        for instance in /tmp/services/${service_name}/*; do
            if [ -f "$instance" ]; then
                cat "$instance"
            fi
        done
    else
        echo "未找到服务: ${service_name}"
    fi
}
# 健康检查
health_check() {
    local host=$1
    local port=$2
    if timeout 2 bash -c "echo > /dev/tcp/${host}/${port}" 2>/dev/null; then
        return 0
    else
        return 1
    fi
}
# 主函数
main() {
    case "$1" in
        register)
            register_service "$2" "$3" "$4"
            ;;
        discover)
            discover_service "$2"
            ;;
        health)
            health_check "$2" "$3"
            ;;
        *)
            echo "用法: $0 {register|discover|health} [参数...]"
            exit 1
            ;;
    esac
}
main "$@"

使用建议

选择合适的方案：
- DNS方案：适合简单场景
- Consul/Eureka：适合微服务架构
- etcd：适合需要强一致性的场景
- Shell脚本：适合快速原型测试
配置健康检查：
- 定期检查服务健康状态
- 自动移除不健康的实例
负载均衡：
- 客户端负载均衡（如Ribbon）
- 服务端负载均衡（如Nginx）
缓存机制：
- 本地缓存服务列表
- 定期刷新缓存
容错处理：
- 熔断机制
- 重试策略
- 降级方案

这些脚本可以根据具体需求进行修改和扩展,实现更适合自己业务场景的服务发现机制。

本文地址： https://www.bmjup.com/post/10112.html