etcd服务发现实践 - 元素码农

发布时间: 2025-04-07 14:10

↑

# etcd服务发现实践

本文将详细介绍如何使用etcd实现服务发现功能，包括基本概念、实现方案和最佳实践。

## 服务发现概述

### 什么是服务发现

服务发现是分布式系统中的一个关键组件，它能够帮助服务消费者自动发现和定位服务提供者的网络位置。在微服务架构中，服务发现尤为重要，因为：

1. 服务实例动态变化
2. 服务地址不固定
3. 需要自动负载均衡
4. 服务健康状态监控

### 为什么选择etcd

etcd作为服务发现系统的优势：

1. 强一致性保证
2. 可靠的Watch机制
3. TTL特性支持
4. 简单的API设计
5. 良好的社区支持

## 实现方案

### 基本架构

### 服务注册

1. 注册服务信息
```go
func registerService(client *clientv3.Client, serviceName, serviceAddr string, ttl int64) error {
    // 创建租约
    lease := clientv3.NewLease(client)
    leaseResp, err := lease.Grant(context.Background(), ttl)
    if err != nil {
        return fmt.Errorf("创建租约失败: %v", err)
    }

// 注册服务
    key := fmt.Sprintf("/services/%s/%s", serviceName, serviceAddr)
    _, err = client.Put(context.Background(), key, serviceAddr, clientv3.WithLease(leaseResp.ID))
    if err != nil {
        return fmt.Errorf("注册服务失败: %v", err)
    }

// 自动续约
    ch, err := client.KeepAlive(context.Background(), leaseResp.ID)
    if err != nil {
        return fmt.Errorf("续约设置失败: %v", err)
    }

go func() {
        for {
            select {
            case _, ok := <-ch:
                if !ok {
                    return
                }
            }
        }
    }()

return nil
}
```

### 服务发现

1. 获取服务列表
```go
func discoverServices(client *clientv3.Client, serviceName string) ([]string, error) {
    prefix := fmt.Sprintf("/services/%s", serviceName)
    resp, err := client.Get(context.Background(), prefix, clientv3.WithPrefix())
    if err != nil {
        return nil, fmt.Errorf("获取服务列表失败: %v", err)
    }

var services []string
    for _, kv := range resp.Kvs {
        services = append(services, string(kv.Value))
    }
    return services, nil
}
```

2. 监听服务变化
```go
func watchServices(client *clientv3.Client, serviceName string) {
    prefix := fmt.Sprintf("/services/%s", serviceName)
    rch := client.Watch(context.Background(), prefix, clientv3.WithPrefix())
    
    for wresp := range rch {
        for _, ev := range wresp.Events {
            switch ev.Type {
            case mvccpb.PUT:
                log.Printf("服务上线: %s", string(ev.Kv.Value))
            case mvccpb.DELETE:
                log.Printf("服务下线: %s", string(ev.Kv.Key))
            }
        }
    }
}
```

## 最佳实践

### 服务健康检查

1. 实现健康检查接口
```go
func healthCheck(service string) bool {
    resp, err := http.Get(fmt.Sprintf("http://%s/health", service))
    if err != nil {
        return false
    }
    defer resp.Body.Close()
    return resp.StatusCode == http.StatusOK
}
```

2. 定期检查服务健康状态
```go
func startHealthCheck(services []string, interval time.Duration) {
    ticker := time.NewTicker(interval)
    for {
        select {
        case <-ticker.C:
            for _, service := range services {
                if !healthCheck(service) {
                    log.Printf("服务不健康: %s", service)
                    // 执行相应的处理逻辑
                }
            }
        }
    }
}
```

### 负载均衡

1. 简单轮询
```go
type RoundRobin struct {
    services []string
    index    int
    mu       sync.Mutex
}

func (r *RoundRobin) Next() string {
    r.mu.Lock()
    defer r.mu.Unlock()
    
    if len(r.services) == 0 {
        return ""
    }
    
    service := r.services[r.index]
    r.index = (r.index + 1) % len(r.services)
    return service
}
```

2. 加权轮询
```go
type WeightedService struct {
    Address string
    Weight  int
}

type WeightedRoundRobin struct {
    services []*WeightedService
    index    int
    mu       sync.Mutex
}

func (w *WeightedRoundRobin) Next() string {
    w.mu.Lock()
    defer w.mu.Unlock()
    
    if len(w.services) == 0 {
        return ""
    }
    
    total := 0
    for _, s := range w.services {
        total += s.Weight
    }
    
    w.index = (w.index + 1) % total
    for _, s := range w.services {
        if w.index < s.Weight {
            return s.Address
        }
        w.index -= s.Weight
    }
    
    return w.services[0].Address
}
```

### 错误处理

1. 重试机制
```go
func retryOperation(operation func() error, maxRetries int, delay time.Duration) error {
    var err error
    for i := 0; i < maxRetries; i++ {
        err = operation()
        if err == nil {
            return nil
        }
        time.Sleep(delay)
    }
    return fmt.Errorf("达到最大重试次数: %v", err)
}
```

2. 服务降级
```go
type ServiceDiscovery struct {
    client    *clientv3.Client
    cache     []string
    cacheLock sync.RWMutex
}

func (sd *ServiceDiscovery) GetServices(serviceName string) ([]string, error) {
    services, err := discoverServices(sd.client, serviceName)
    if err != nil {
        // 服务发现失败时使用缓存
        sd.cacheLock.RLock()
        defer sd.cacheLock.RUnlock()
        return sd.cache, nil
    }
    
    // 更新缓存
    sd.cacheLock.Lock()
    sd.cache = services
    sd.cacheLock.Unlock()
    
    return services, nil
}
```

## 应用示例

### 完整的服务注册发现示例

```go
package main

import (
    "context"
    "fmt"
    "log"
    "time"
    "go.etcd.io/etcd/clientv3"
)

type ServiceRegistry struct {
    client     *clientv3.Client
    serviceName string
    serviceAddr string
    leaseID     clientv3.LeaseID
}

func NewServiceRegistry(endpoints []string, serviceName, serviceAddr string) (*ServiceRegistry, error) {
    client, err := clientv3.New(clientv3.Config{
        Endpoints:   endpoints,
        DialTimeout: 5 * time.Second,
    })
    if err != nil {
        return nil, err
    }

return &ServiceRegistry{
        client:     client,
        serviceName: serviceName,
        serviceAddr: serviceAddr,
    }, nil
}

func (sr *ServiceRegistry) Register() error {
    // 创建租约
    lease := clientv3.NewLease(sr.client)
    leaseResp, err := lease.Grant(context.Background(), 10)
    if err != nil {
        return err
    }
    sr.leaseID = leaseResp.ID

// 注册服务
    key := fmt.Sprintf("/services/%s/%s", sr.serviceName, sr.serviceAddr)
    _, err = sr.client.Put(context.Background(), key, sr.serviceAddr, clientv3.WithLease(sr.leaseID))
    if err != nil {
        return err
    }

// 自动续约
    ch, err := sr.client.KeepAlive(context.Background(), sr.leaseID)
    if err != nil {
        return err
    }

go func() {
        for {
            select {
            case _, ok := <-ch:
                if !ok {
                    log.Println("服务租约已过期")
                    return
                }
            }
        }
    }()

return nil
}

func (sr *ServiceRegistry) Unregister() error {
    if sr.leaseID != 0 {
        _, err := sr.client.Revoke(context.Background(), sr.leaseID)
        return err
    }
    return nil
}

func main() {
    // 创建服务注册器
    registry, err := NewServiceRegistry(
        []string{"localhost:2379"},
        "my-service",
        "localhost:8080",
    )
    if err != nil {
        log.Fatal(err)
    }

// 注册服务
    err = registry.Register()
    if err != nil {
        log.Fatal(err)
    }

// 监听服务变化
    go watchServices(registry.client, "my-service")

// 等待信号
    select {}
}
```

## 注意事项

1. 租约设置
   - TTL不要设置太短，避免频繁续约
   - 建议设置为10-30秒

2. 监视操作
   - 处理Watch事件要及时，避免阻塞
   - 考虑使用缓冲通道

3. 错误处理
   - 实现优雅降级机制
   - 使用本地缓存作为备份

4. 性能优化
   - 合理使用批量操作
   - 避免频繁创建/关闭客户端

## 总结

etcd作为服务发现系统具有以下优点：

1. 强一致性保证服务信息的可靠性
2. Watch机制支持实时服务状态更新
3. TTL机制自动清理失效服务
4. 简单易用的API降低开发成本

在实际应用中，需要注意：

1. 合理设计服务注册的数据结构
2. 实现可靠的健康检查机制
3. 做好容错和降级处理
4. 优化性能和资源使用