IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

脚本如何发现调用可用服务

wen 实用脚本 28

本文目录导读：

脚本如何发现调用可用服务

  1. 模式 1：基于 DNS 的服务发现（最通用、简单）
  2. 模式 2：基于 HTTP API 的注册中心（如 Consul, Eureka, Nacos）
  3. 模式 3：基于 K8s API（Kubernetes 环境）
  4. 模式 4：自定义配置文件 / 环境变量（静态发现）
  5. 脚本发现服务的核心流程总结
  6. 注意事项

这是一个很典型的服务发现（Service Discovery）问题，脚本发现可用服务的具体方法取决于你的架构环境（是微服务、单体、还是云原生）以及服务注册的方式。

下面从 4 种最常见的模式 来解答，并给出相应的脚本实现思路和示例代码（主要用 Shell 和 Python）。

模式 1：基于 DNS 的服务发现（最通用、简单）

这是云原生（Kubernetes）和传统分布式系统中最常用的方式，服务注册在 DNS 服务器中，脚本通过解析域名或 SRV 记录找到可用服务。

适用场景： Kubernetes Service、Consul DNS、CoreDNS、Mesos DNS。

脚本做法：

  • 查询 A 记录（获取 IP 列表）。
  • 查询 SRV 记录（获取 IP + 端口 + 权重）。

Shell 示例（用 dig 命令）：

#!/bin/bash
# 发现名为 "my-service" 的后端所有可用 IP（假设在 k8s 中）
SERVICE_NAME="my-service.namespace.svc.cluster.local"
# 解析 A 记录，获取所有 IP
AVAILABLE_IPS=$(dig +short $SERVICE_NAME)
if [ -z "$AVAILABLE_IPS" ]; then
    echo "错误：未发现任何可用服务实例"
    exit 1
fi
echo "发现可用服务实例 IP："
echo "$AVAILABLE_IPS"
# 假设使用第一个 IP 访问
FIRST_IP=$(echo "$AVAILABLE_IPS" | head -n 1)
curl http://$FIRST_IP:8080/health

Python 示例（用 socket 库）：

import socket
def discover_service_by_dns(service_name, port=None):
    try:
        # 获取全部 IP (A 记录)
        ips = socket.getaddrinfo(service_name, port or 80)
        instances = []
        for info in ips:
            # socket.getaddrinfo 返回多个协议族，去重 IP
            ip = info[4][0]
            if ip not in [i['ip'] for i in instances]:
                instances.append({'ip': ip, 'port': port or 80})
        return instances
    except socket.gaierror:
        return []
# 使用示例
instances = discover_service_by_dns('my-service.my-namespace.svc.cluster.local', 8080)
if instances:
    print(f"发现 {len(instances)} 个实例: {instances}")
    # 随机选一个
    import random
    target = random.choice(instances)
    print(f"选中: {target['ip']}:{target['port']}")
else:
    print("无可用服务")

模式 2：基于 HTTP API 的注册中心（如 Consul, Eureka, Nacos）

适用场景： Consul、Eureka、Nacos、etcd。

脚本做法： 脚本向注册中心的 HTTP API 发送请求，解析返回的 JSON/XML，提取健康实例的 IP 和端口。

Consul 示例（Shell + jq）： 假设 Consul Agent 运行在本地 8500 端口。

#!/bin/bash
SERVICE_NAME="my-api"
# 查询 Consul 健康检查通过的实例
HEALTHY_INSTANCES=$(curl -s http://localhost:8500/v1/health/service/$SERVICE_NAME?passing=true | jq -r '.[] | select(.Checks[].Status == "passing") | .Service.Address + ":" + (.Service.Port | tostring)')
if [ -z "$HEALTHY_INSTANCES" ]; then
    echo "Consul 中未发现健康实例"
    exit 1
fi
echo "发现健康实例："
echo "$HEALTHY_INSTANCES"
# 随机选取一个实例
TARGET=$(echo "$HEALTHY_INSTANCES" | shuf -n 1)
echo "调用实例: $TARGET"
curl http://$TARGET/api/v1/status

Python（用 Nacos API 示例）：

import requests
import random
def discover_nacos(server_url, service_name, namespace='public'):
    url = f"{server_url}/nacos/v1/ns/instance/list"
    params = {
        'serviceName': service_name,
        'namespaceId': namespace,
        'healthyOnly': True
    }
    resp = requests.get(url, params=params, timeout=5)
    if resp.status_code == 200:
        data = resp.json()
        instances = data.get('hosts', [])
        if instances:
            return [(inst['ip'], inst['port']) for inst in instances if inst['healthy']]
    return []
# 使用
instances = discover_nacos('http://192.168.1.100:8848', 'order-service')
if instances:
    ip, port = random.choice(instances)
    print(f"调用 {ip}:{port}/health")

模式 3：基于 K8s API（Kubernetes 环境）

适用场景： 脚本运行在 Pod 内部或拥有 K8s 集群访问权限。

脚本做法： 利用 kubectl 命令或 K8s Python 客户端，根据标签选择器（Selector）查询 Running 状态的 Pod IP。

Shell（使用 kubectl 和 jq）：

#!/bin/bash
# 假设脚本运行在 Pod 内，且有 serviceaccount 权限
LABEL_SELECTOR="app=my-service,version=v1"
NAMESPACE="production"
# 获取所有 Ready 的 Pod IP
POD_IPS=$(kubectl get pods -n $NAMESPACE -l $LABEL_SELECTOR -o json | jq -r '.items[] | select(.status.phase == "Running") | .status.podIP')
if [ -z "$POD_IPS" ]; then
    echo "未发现运行中的 Pod"
    exit 1
fi
echo "可用 Pod IP:"
echo "$POD_IPS"
# 选择第一个调用
curl http://$(echo "$POD_IPS" | head -1):8080/health

Python（用 kubernetes 客户端）：

from kubernetes import client, config
def discover_k8s_pods(namespace, label_selector):
    # 加载配置（Pod内自动使用 ServiceAccount；本地使用 ~/.kube/config）
    config.load_incluster_config() # 或 config.load_kube_config()
    v1 = client.CoreV1Api()
    ret = v1.list_namespaced_pod(namespace, label_selector=label_selector)
    instances = []
    for pod in ret.items:
        if pod.status.phase == 'Running':
            instances.append(pod.status.pod_ip)
    return instances
# 使用
pods = discover_k8s_pods('production', 'app=order-service')
if pods:
    print(f"发现 Pod IPs: {pods}")

模式 4：自定义配置文件 / 环境变量（静态发现）

适用场景： 开发环境、固定集群、或简单的服务。

脚本做法： 读取配置文件（YAML/JSON）或环境变量中预定义的服务列表,并尝试发起健康检查以过滤出可用服务。

Shell 示例：

#!/bin/bash
# 从环境变量读取服务列表（格式: ip1:port1,ip2:port2）
SERVICES="${MY_SERVICE_LIST:-192.168.1.10:8080,192.168.1.11:8080}"
IFS=',' read -ra ADDR <<< "$SERVICES"
for addr in "${ADDR[@]}"; do
    # 尝试 TCP 连接或 HTTP 健康检查
    if nc -z -w 2 ${addr//:/ }; then
        echo "发现可用服务: $addr"
        HEALTHY_LIST+=("$addr")
    fi
done
if [ ${#HEALTHY_LIST[@]} -eq 0 ]; then
    echo "无可用服务"
    exit 1
fi
# 调用第一个
curl http://${HEALTHY_LIST[0]}/api

脚本发现服务的核心流程总结

无论哪种模式，一个完备的“发现”脚本通常分 3 步：

  1. 查询 / 拉取：从注册中心、DNS、API Server 或配置处获取所有备选实例列表（IP:Port）。
  2. 过滤 / 健康检查：
    • 利用注册中心自带的健康状态（如 Consul 的 ?passing=true）。
    • 或手动对每个实例发一个简单的 TCP 连接或 HTTP /health 请求。
  3. 选择 / 负载均衡：
    • 随机（最常见）。
    • 轮询（维护一个计数器）。
    • 最小连接（需要额外逻辑）。

注意事项

  • 缓存：不要每次都查询注册中心，避免造成压力（建议缓存 TTL 5-30 秒）。
  • 容错：如果查询注册中心失败，应使用本地缓存的上一轮有效实例,而不是直接报错。
  • 网络策略：确保脚本所在的环境能访问到注册中心或 DNS 服务器。

一句话总结： 如果你的服务管理是标准的，优先用 Consul/Nacos/etcd 的 HTTP API 或 K8s 原生 API；如果是简单的固定集群，用 DNS SRV 记录加上自定义健康检查更轻量，脚本只要做到“查列表 -> 过滤 -> 调用”,就是一套完整的服务发现流程。

上一篇如何编写分布式任务脚本

下一篇脚本怎样检测剔除故障节点

相关文章

抱歉，评论功能暂时关闭!