服务发现与注册中心

简介

在微服务架构中，服务实例动态变化，服务发现机制让消费者无需硬编码服务地址。理解客户端发现、服务端发现的模式差异，以及 Consul 等注册中心的集成方式，有助于构建弹性的服务通信。

特点

• 1.发现模式 — 客户端发现 vs 服务端发现
• 2.Consul 集成 — 注册、发现、健康检查
• 3.负载均衡 — 客户端负载均衡策略
• 4.K8s 服务 — Kubernetes 原生服务发现
• 5.故障恢复 — 缓存与降级策略

服务发现模式

客户端发现 vs 服务端发现

// 客户端发现模式：
// 消费者 → 查询注册中心 → 获取服务实例列表 → 选择实例直接调用
// 优点：无中间代理，延迟低
// 缺点：客户端复杂度高，需要实现负载均衡

// 服务端发现模式（API Gateway / Load Balancer）：
// 消费者 → API Gateway → 查询注册中心 → 转发到服务实例
// 优点：客户端简单
// 缺点：增加一跳，网关成为瓶颈

// Kubernetes 服务发现：
// ClusterIP Service → kube-proxy → iptables/IPVS → Pod
// 客户端只需访问 Service DNS 名称

// DNS 服务发现（最简单）：
// order-service → DNS 解析 → 返回 IP 地址
// 适用于 K8s Service / Docker Compose

Consul 集成

服务注册与发现

// dotnet add package Consul

// 1. 服务注册配置
public class ConsulRegistrationOptions
{
    public string ServiceName { get; set; } = "";
    public string ServiceId { get; set; } = Guid.NewGuid().ToString("N")[..8];
    public string Address { get; set; } = "localhost";
    public int Port { get; set; } = 5000;
    public string[] Tags { get; set; } = Array.Empty<string>();
    public TimeSpan HealthCheckInterval { get; set; } = TimeSpan.FromSeconds(10);
    public TimeSpan DeregisterCriticalAfter { get; set; } = TimeSpan.FromMinutes(1);
}

// 2. 自动注册中间件
public class ConsulRegistrationMiddleware
{
    private readonly IConsulClient _consulClient;
    private readonly ConsulRegistrationOptions _options;
    private readonly ILogger<ConsulRegistrationMiddleware> _logger;

    public ConsulRegistrationMiddleware(
        IConsulClient consulClient,
        IOptions<ConsulRegistrationOptions> options,
        ILogger<ConsulRegistrationMiddleware> logger)
    {
        _consulClient = consulClient;
        _options = options.Value;
        _logger = logger;
    }

    public async Task RegisterAsync()
    {
        var registration = new AgentServiceRegistration
        {
            ID = _options.ServiceId,
            Name = _options.ServiceName,
            Address = _options.Address,
            Port = _options.Port,
            Tags = _options.Tags,

            Check = new AgentServiceCheck
            {
                HTTP = $"http://{_options.Address}:{_options.Port}/health",
                Interval = _options.HealthCheckInterval,
                Timeout = TimeSpan.FromSeconds(5),
                DeregisterCriticalServiceAfter = _options.DeregisterCriticalAfter
            }
        };

        await _consulClient.Agent.ServiceRegister(registration);
        _logger.LogInformation(
            "服务已注册: {Name} ({Id}) at {Address}:{Port}",
            _options.ServiceName, _options.ServiceId, _options.Address, _options.Port);
    }

    public async Task DeregisterAsync()
    {
        await _consulClient.Agent.ServiceDeregister(_options.ServiceId);
        _logger.LogInformation("服务已注销: {Id}", _options.ServiceId);
    }
}

// 3. 注册到 ASP.NET Core 生命周期
public static class ConsulExtensions
{
    public static IApplicationBuilder UseConsulRegistration(this IApplicationBuilder app)
    {
        var lifetime = app.ApplicationServices.GetRequiredService<IHostApplicationLifetime>();
        var registration = app.ApplicationServices.GetRequiredService<ConsulRegistrationMiddleware>();

        lifetime.ApplicationStarted.Register(async () =>
        {
            await registration.RegisterAsync();
        });

        lifetime.ApplicationStopping.Register(async () =>
        {
            await registration.DeregisterAsync();
        });

        return app;
    }
}

// 4. DI 注册
builder.Services.AddSingleton<IConsulClient>(sp => new ConsulClient(config =>
{
    config.Address = new Uri(builder.Configuration["Consul:Address"] ?? "http://localhost:8500");
}));

builder.Services.Configure<ConsulRegistrationOptions>(builder.Configuration.GetSection("Consul:Registration"));
builder.Services.AddSingleton<ConsulRegistrationMiddleware>();

// 5. 启用
app.UseConsulRegistration();

服务发现客户端

// 服务发现 + 客户端负载均衡
public class ServiceDiscoveryClient
{
    private readonly IConsulClient _consulClient;
    private readonly ILogger<ServiceDiscoveryClient> _logger;
    private readonly ConcurrentDictionary<string, ServiceCache> _cache = new();
    private readonly Random _random = new();

    public ServiceDiscoveryClient(IConsulClient consulClient, ILogger<ServiceDiscoveryClient> logger)
    {
        _consulClient = consulClient;
        _logger = logger;
    }

    // 发现服务实例
    public async Task<ServiceInstance?> ResolveAsync(string serviceName, CancellationToken ct = default)
    {
        var instances = await GetInstancesAsync(serviceName, ct);

        if (instances.Count == 0)
        {
            _logger.LogWarning("没有可用的 {ServiceName} 实例", serviceName);
            return null;
        }

        // 随机负载均衡
        var index = _random.Next(instances.Count);
        return instances[index];
    }

    // 获取所有健康实例
    private async Task<List<ServiceInstance>> GetInstancesAsync(string serviceName, CancellationToken ct)
    {
        // 检查缓存
        if (_cache.TryGetValue(serviceName, out var cache) && !cache.IsExpired)
        {
            return cache.Instances;
        }

        // 查询 Consul
        var response = await _consulClient.Health.Service(serviceName, tag: "", passingOnly: true, ct: ct);

        var instances = response.Response
            .Select(s => new ServiceInstance(
                s.Service.ID,
                s.Service.Address,
                s.Service.Port,
                s.Service.Tags ?? Array.Empty<string>()))
            .ToList();

        // 更新缓存（5 秒）
        _cache[serviceName] = new ServiceCache(instances, TimeSpan.FromSeconds(5));

        _logger.LogDebug("发现 {ServiceName} 的 {Count} 个实例", serviceName, instances.Count);
        return instances;
    }
}

public record ServiceInstance(string Id, string Address, int Port, string[] Tags)
{
    public string BaseUrl => $"http://{Address}:{Port}";
}

public record ServiceCache(List<ServiceInstance> Instances, TimeSpan TTL)
{
    private readonly DateTime _createdAt = DateTime.UtcNow;
    public bool IsExpired => DateTime.UtcNow - _createdAt > TTL;
}

// 与 HttpClientFactory 集成
public class ConsulHttpMessageHandler : DelegatingHandler
{
    private readonly ServiceDiscoveryClient _discovery;
    private readonly string _serviceName;
    private readonly ILogger<ConsulHttpMessageHandler> _logger;

    public ConsulHttpMessageHandler(
        ServiceDiscoveryClient discovery,
        string serviceName,
        ILogger<ConsulHttpMessageHandler> logger)
    {
        _discovery = discovery;
        _serviceName = serviceName;
        _logger = logger;
    }

    protected override async Task<HttpResponseMessage> SendAsync(
        HttpRequestMessage request, CancellationToken cancellationToken)
    {
        // 发现服务实例
        var instance = await _discovery.ResolveAsync(_serviceName, cancellationToken);

        if (instance == null)
        {
            throw new InvalidOperationException($"没有可用的 {_serviceName} 实例");
        }

        // 替换请求 URL
        var originalUri = request.RequestUri!;
        var newUri = new UriBuilder(originalUri)
        {
            Host = instance.Address,
            Port = instance.Port
        }.Uri;

        request.RequestUri = newUri;

        _logger.LogDebug("转发到 {Service}: {Url}", _serviceName, newUri);

        return await base.SendAsync(request, cancellationToken);
    }
}

// 注册
builder.Services.AddSingleton<ServiceDiscoveryClient>();
builder.Services.AddHttpClient("OrderService")
    .ConfigurePrimaryHttpMessageHandler(sp =>
    {
        var discovery = sp.GetRequiredService<ServiceDiscoveryClient>();
        var logger = sp.GetRequiredService<ILogger<ConsulHttpMessageHandler>>();
        return new ConsulHttpMessageHandler(discovery, "order-service", logger);
    });

负载均衡策略

多种策略实现

// 负载均衡策略接口
public interface ILoadBalancer
{
    ServiceInstance Select(List<ServiceInstance> instances, string? key = null);
}

// 随机策略
public class RandomLoadBalancer : ILoadBalancer
{
    private readonly Random _random = new();

    public ServiceInstance Select(List<ServiceInstance> instances, string? key = null)
        => instances[_random.Next(instances.Count)];
}

// 轮询策略
public class RoundRobinLoadBalancer : ILoadBalancer
{
    private int _index;
    private readonly object _lock = new();

    public ServiceInstance Select(List<ServiceInstance> instances, string? key = null)
    {
        lock (_lock)
        {
            _index = (_index + 1) % instances.Count;
            return instances[_index];
        }
    }
}

// 加权轮询策略
public class WeightedRoundRobinBalancer : ILoadBalancer
{
    private readonly Dictionary<string, int> _weights;
    private int _currentWeight;
    private int _currentIndex;
    private readonly object _lock = new();

    public WeightedRoundRobinBalancer(Dictionary<string, int> weights)
    {
        _weights = weights;
    }

    public ServiceInstance Select(List<ServiceInstance> instances, string? key = null)
    {
        lock (_lock)
        {
            // 平滑加权轮询
            var total = instances.Sum(i => _weights.GetValueOrDefault(i.Id, 1));

            while (true)
            {
                _currentIndex = (_currentIndex + 1) % instances.Count;
                if (_currentIndex == 0)
                {
                    _currentWeight -= total;
                    if (_currentWeight <= 0)
                        _currentWeight = _weights.Values.Max();
                }

                var instance = instances[_currentIndex];
                var weight = _weights.GetValueOrDefault(instance.Id, 1);
                if (weight >= _currentWeight)
                    return instance;
            }
        }
    }
}

// 一致性哈希策略
public class ConsistentHashBalancer : ILoadBalancer
{
    private readonly SortedDictionary<uint, ServiceInstance> _ring = new();

    public ConsistentHashBalancer(List<ServiceInstance> instances, int virtualNodes = 150)
    {
        foreach (var instance in instances)
        {
            for (int i = 0; i < virtualNodes; i++)
            {
                var hash = Hash($"{instance.Id}:{i}");
                _ring[hash] = instance;
            }
        }
    }

    public ServiceInstance Select(List<ServiceInstance> instances, string? key = null)
    {
        if (string.IsNullOrEmpty(key)) key = Guid.NewGuid().ToString();
        var hash = Hash(key);

        // 顺时针找到第一个节点
        var node = _ring.FirstOrDefault(kvp => kvp.Key >= hash);
        return node.Value ?? _ring.First().Value;
    }

    private static uint Hash(string key)
    {
        var bytes = MD5.HashData(Encoding.UTF8.GetBytes(key));
        return BitConverter.ToUInt32(bytes, 0);
    }
}

Kubernetes 原生服务发现

K8s Service 与 DNS

# Kubernetes 服务发现机制
# 1. ClusterIP Service — 内部负载均衡
apiVersion: v1
kind: Service
metadata:
  name: order-service
  namespace: production
spec:
  type: ClusterIP
  selector:
    app: order-service
  ports:
    - port: 80
      targetPort: 8080

---
# 2. Headless Service — 直接返回 Pod IP（用于有状态服务）
apiVersion: v1
kind: Service
metadata:
  name: database
spec:
  clusterIP: None  # Headless
  selector:
    app: postgres
  ports:
    - port: 5432

---
# 3. ExternalName Service — 外部服务别名
apiVersion: v1
kind: Service
metadata:
  name: external-api
spec:
  type: ExternalName
  externalName: api.external-provider.com

---
# DNS 解析规则：
# 同命名空间：order-service
# 跨命名空间：order-service.production.svc.cluster.local
# Headless Pod：database-0.database.default.svc.cluster.local

ASP.NET Core 使用 K8s 服务发现

// 在 K8s 中运行时，直接使用 Service DNS 名称
// 无需额外的注册中心

// appsettings.json
// {
//   "ServiceUrls": {
//     "OrderService": "http://order-service:80",
//     "ProductService": "http://product-service:80",
//     "Database": "database-0.database.default:5432"
//   }
// }

// 通过 HttpClientFactory 调用
builder.Services.AddHttpClient("OrderService", client =>
{
    client.BaseAddress = new Uri(
        builder.Configuration["ServiceUrls:OrderService"] 
        ?? "http://order-service:80");
    client.Timeout = TimeSpan.FromSeconds(30);
})
.ConfigurePrimaryHttpMessageHandler(() => new SocketsHttpHandler
{
    PooledConnectionLifetime = TimeSpan.FromMinutes(5) // 定期刷新 DNS
});

// 关键：K8s Service 的 DNS 解析结果会随 Pod 变化而更新
// 设置 PooledConnectionLifetime 确保定期重新解析 DNS

服务健康检查

多层健康检查策略

// 健康检查端点 — 供注册中心和负载均衡器探测
builder.Services.AddHealthChecks()
    .AddCheck("self", () => HealthCheckResult.Healthy())
    .AddNpgSql(builder.Configuration.GetConnectionString("Default")!)
    .AddRedis(builder.Configuration.GetConnectionString("Redis")!)
    .AddRabbitMQ(builder.Configuration.GetConnectionString("RabbitMQ")!)
    .AddUrlGroup(new Uri("https://external-api.example.com/health"), "external-api");

app.MapHealthChecks("/health", new HealthCheckOptions
{
    ResponseWriter = async (context, report) =>
    {
        var result = JsonSerializer.Serialize(new
        {
            status = report.Status.ToString(),
            checks = report.Entries.Select(e => new
            {
                name = e.Key,
                status = e.Value.Status.ToString(),
                duration = e.Value.Duration.TotalMilliseconds,
                description = e.Value.Description
            }),
            totalDuration = report.TotalDuration.TotalMilliseconds
        });
        context.Response.ContentType = "application/json";
        await context.Response.WriteAsync(result);
    }
});

// Consul 健康检查配置
var registration = new AgentServiceRegistration
{
    ID = serviceId,
    Name = serviceName,
    Address = address,
    Port = port,
    Check = new AgentServiceCheck
    {
        HTTP = $"http://{address}:{port}/health",
        Interval = TimeSpan.FromSeconds(10),
        Timeout = TimeSpan.FromSeconds(5),
        DeregisterCriticalServiceAfter = TimeSpan.FromMinutes(1),
        // TLS 跳过（开发环境）
        TLSSkipVerify = true
    }
};

K8s 探针配置

# Kubernetes Pod 探针配置
apiVersion: apps/v1
kind: Deployment
metadata:
  name: api-service
spec:
  replicas: 3
  template:
    spec:
      containers:
        - name: api
          image: myapp:v1.0
          ports:
            - containerPort: 8080
          # 存活探针 — 失败时重启容器
          livenessProbe:
            httpGet:
              path: /health
              port: 8080
            initialDelaySeconds: 15
            periodSeconds: 20
            timeoutSeconds: 5
            failureThreshold: 3
          # 就绪探针 — 失败时从 Service 移除
          readinessProbe:
            httpGet:
              path: /health/ready
              port: 8080
            initialDelaySeconds: 5
            periodSeconds: 10
            timeoutSeconds: 3
            failureThreshold: 3
          # 启动探针 — 应用启动完成前阻止其他探针
          startupProbe:
            httpGet:
              path: /health
              port: 8080
            failureThreshold: 30
            periodSeconds: 10

服务发现容错

客户端缓存与降级

// 带缓存的服务发现 — 防止注册中心不可用时服务全部失败
public class ResilientServiceDiscovery
{
    private readonly ServiceDiscoveryClient _discovery;
    private readonly ConcurrentDictionary<string, List<ServiceInstance>> _fallbackCache = new();
    private readonly ILogger<ResilientServiceDiscovery> _logger;

    public async Task<ServiceInstance?> ResolveAsync(string serviceName, CancellationToken ct)
    {
        try
        {
            var instance = await _discovery.ResolveAsync(serviceName, ct);
            
            if (instance != null)
            {
                // 更新本地缓存
                var instances = await _discovery.GetAllInstancesAsync(serviceName, ct);
                _fallbackCache[serviceName] = instances;
                return instance;
            }
            
            // 发现服务返回 null，使用缓存
            return GetFromCache(serviceName);
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex, "服务发现失败，使用本地缓存: {ServiceName}", serviceName);
            return GetFromCache(serviceName);
        }
    }

    private ServiceInstance? GetFromCache(string serviceName)
    {
        if (_fallbackCache.TryGetValue(serviceName, out var instances) && instances.Count > 0)
        {
            _logger.LogInformation("使用缓存的服务实例: {ServiceName} ({Count} 个)", 
                serviceName, instances.Count);
            var random = new Random();
            return instances[random.Next(instances.Count)];
        }
        return null;
    }
}

优点

• 1.动态发现 — 服务实例自动注册和发现
• 2.健康检查 — 自动移除不健康实例
• 3.负载均衡 — 多种策略可选
• 4.配置中心 — Consul 支持 KV 配置

缺点

• 1.注册中心依赖 — 单点故障风险
• 2.一致性延迟 — 注册信息有传播延迟
• 3.调试复杂 — 服务地址动态变化
• 4.额外基础设施 — 需要部署和维护注册中心

总结

服务发现分为客户端发现（消费者查询注册中心直接调用）和服务端发现（通过网关转发）两种模式。Consul 提供服务注册、健康检查和 KV 配置功能，通过 IHostApplicationLifetime 实现自动注册和注销。客户端负载均衡支持随机、轮询、加权轮询和一致性哈希等策略。在 Kubernetes 环境中，推荐使用原生 Service DNS 发现，无需额外注册中心。

关键知识点

• 先分清这个主题位于请求链路、后台任务链路还是基础设施链路。
• 服务端主题通常不只关心功能正确，还关心稳定性、性能和可观测性。
• 任何框架能力都要结合配置、生命周期、异常传播和外部依赖一起看。
• Kubernetes 主题必须同时看资源对象、调度行为、网络暴露和配置分发。

项目落地视角

• 画清请求进入、业务执行、外部调用、日志记录和错误返回的完整路径。
• 为关键链路补齐超时、重试、熔断、追踪和结构化日志。
• 把配置与敏感信息分离，并明确不同环境的差异来源。
• 上线前检查镜像、命名空间、探针、资源限制、Service/Ingress 和配置来源。

常见误区

• 只会堆中间件或组件，不知道它们在链路中的执行顺序。
• 忽略生命周期和线程池、连接池等运行时资源约束。
• 没有监控和测试就对性能或可靠性下结论。
• 只会 apply YAML，不理解对象之间的依赖关系。

进阶路线

• 继续向运行时行为、可观测性、发布治理和微服务协同深入。
• 把主题和数据库、缓存、消息队列、认证授权联动起来理解。
• 沉淀团队级模板，包括统一异常处理、配置约定和基础设施封装。
• 继续补齐调度、网络策略、存储、GitOps 和平台工程能力。

适用场景

• 当你准备把《服务发现与注册中心》真正落到项目里时，最适合先在一个独立模块或最小样例里验证关键路径。
• 适合 API 服务、后台任务、实时通信、认证授权和微服务协作场景。
• 当需求开始涉及稳定性、性能、可观测性和发布流程时，这类主题会成为基础设施能力。

落地建议

• 先定义请求链路与失败路径，再决定中间件、过滤器、服务边界和依赖方式。
• 为关键链路补日志、指标、追踪、超时与重试策略。
• 环境配置与敏感信息分离，避免把生产参数写死在代码或镜像里。

排错清单

• 先确认问题发生在路由、模型绑定、中间件、业务层还是基础设施层。
• 检查 DI 生命周期、配置来源、序列化规则和认证上下文。
• 查看线程池、连接池、缓存命中率和外部依赖超时。

复盘问题

• 如果把《服务发现与注册中心》放进你的当前项目，最先要验证的输入、输出和失败路径分别是什么？
• 《服务发现与注册中心》最容易在什么规模、什么边界条件下暴露问题？你会用什么指标或日志去确认？
• 相比默认实现或替代方案，采用《服务发现与注册中心》最大的收益和代价分别是什么？

延伸阅读

• 后台任务 Background Service
• 序列化性能对比与调优
• 安全编码实践
• 流式响应与 Server-Sent Events