8.4 服务发现与注册机制¶

package main

import (
    "context"
    "fmt"
    "log"
    "time"

    "github.com/hashicorp/consul/api"
)

type ConsulClient struct {
    client *api.Client
}

func NewConsulClient(addr string) (*ConsulClient, error) {
    config := api.DefaultConfig()
    config.Address = addr
    client, err := api.NewClient(config)
    if err != nil {
        return nil, err
    }
    return &ConsulClient{client: client}, nil
}

// RegisterService 注册服务到Consul
func (c *ConsulClient) RegisterService(serviceID, serviceName, address string, port int) error {
    registration := &api.AgentServiceRegistration{
        ID:      serviceID,
        Name:    serviceName,
        Address: address,
        Port:    port,
        Check: &api.AgentServiceCheck{
            HTTP:                           fmt.Sprintf("http://%s:%d/health", address, port), // 健康检查端点
            Interval:                       "10s",                                           // 检查间隔
            Timeout:                        "5s",                                            // 检查超时
            DeregisterCriticalServiceAfter: "30s",                                           // 故障超时注销时间
        },
    }
    return c.client.Agent().ServiceRegister(registration)
}

// DiscoverServices 从Consul发现服务
func (c *ConsulClient) DiscoverServices(serviceName string) ([]*api.ServiceEntry, error) {
    entries, _, err := c.client.Health().Service(serviceName, "", true, nil)
    return entries, err
}

// DeregisterService 注销服务
func (c *ConsulClient) DeregisterService(serviceID string) error {
    return c.client.Agent().ServiceDeregister(serviceID)
}

func main() {
    // 1. 连接至Consul Agent (假设运行在本地8500端口)
    consulClient, err := NewConsulClient("localhost:8500")
    if err != nil {
        log.Fatalf("Failed to connect to Consul: %v", err)
    }

    // 2. 注册一个示例服务
    serviceID := "my-web-app-1"
    serviceName := "web-app"
    err = consulClient.RegisterService(serviceID, serviceName, "localhost", 8080)
    if err != nil {
        log.Fatalf("Failed to register service: %v", err)
    }
    fmt.Printf("Service %s registered successfully.\n", serviceID)

    // 3. 模拟服务运行...
    time.Sleep(2 * time.Second)

    // 4. 发现所有健康的 "web-app" 服务实例
    entries, err := consulClient.DiscoverServices(serviceName)
    if err != nil {
        log.Fatalf("Failed to discover services: %v", err)
    }
    fmt.Printf("Discovered %d healthy instances of %s:\n", len(entries), serviceName)
    for _, entry := range entries {
        fmt.Printf("  - %s:%d\n", entry.Service.Address, entry.Service.Port)
    }

    // 5. 程序退出前注销服务
    defer func() {
        err := consulClient.DeregisterService(serviceID)
        if err != nil {
            log.Printf("Failed to deregister service: %v", err)
        } else {
            fmt.Printf("Service %s deregistered.\n", serviceID)
        }
    }()

    // 保持运行一段时间以便观察
    time.Sleep(30 * time.Second)
}

package main

import (
    "context"
    "fmt"
    "log"
    "time"

    clientv3 "go.etcd.io/etcd/client/v3"
)

type EtcdServiceDiscovery struct {
    client *clientv3.Client
}

func NewEtcdServiceDiscovery(endpoints []string) (*EtcdServiceDiscovery, error) {
    cli, err := clientv3.New(clientv3.Config{
        Endpoints:   endpoints,
        DialTimeout: 5 * time.Second,
    })
    if err != nil {
        return nil, err
    }
    return &EtcdServiceDiscovery{client: cli}, nil
}

// RegisterService 使用租约注册服务
func (sd *EtcdServiceDiscovery) RegisterService(serviceKey, serviceVal string, ttl int64) error {
    // 创建租约
    leaseResp, err := sd.client.Grant(context.TODO(), ttl)
    if err != nil {
        return err
    }

    // 将键值对与租期绑定
    _, err = sd.client.Put(context.TODO(), serviceKey, serviceVal, clientv3.WithLease(leaseResp.ID))
    if err != nil {
        return err
    }

    // 定期续租以确保服务不被删除
    keepAliveCh, err := sd.client.KeepAlive(context.TODO(), leaseResp.ID)
    if err != nil {
        return err
    }
    // 处理续租响应通道，防止通道堵塞（通常需要在一个goroutine中处理）
    go func() {
        for range keepAliveCh {
            // 续租成功，可以打印日志或进行其他操作
        }
    }()
    return nil
}

// DiscoverServices 发现服务并监听变化
func (sd *EtcdServiceDiscovery) DiscoverServices(servicePrefix string) {
    // 首先获取当前的所有服务
    getResp, err := sd.client.Get(context.TODO(), servicePrefix, clientv3.WithPrefix())
    if err != nil {
        log.Printf("Failed to get initial services: %v", err)
    } else {
        fmt.Println("Initial services:")
        for _, kv := range getResp.Kvs {
            fmt.Printf("  Key: %s, Value: %s\n", string(kv.Key), string(kv.Value))
        }
    }

    // 监听前缀的变化
    watchChan := sd.client.Watch(context.TODO(), servicePrefix, clientv3.WithPrefix())
    fmt.Printf("Watching for changes on prefix: %s...\n", servicePrefix)
    for watchResp := range watchChan {
        for _, event := range watchResp.Events {
            switch event.Type {
            case clientv3.EventTypePut:
                fmt.Printf("Service added/updated: Key=%s, Value=%s\n", string(event.Kv.Key), string(event.Kv.Value))
            case clientv3.EventTypeDelete:
                fmt.Printf("Service deleted: Key=%s\n", string(event.Kv.Key))
            }
        }
    }
}

func (sd *EtcdServiceDiscovery) Close() error {
    return sd.client.Close()
}

func main() {
    // 1. 连接etcd集群
    endpoints := []string{"localhost:2379"}
    sd, err := NewEtcdServiceDiscovery(endpoints)
    if err != nil {
        log.Fatalf("Failed to connect to etcd: %v", err)
    }
    defer sd.Close()

    servicePrefix := "/registry/services/web-app/"
    serviceKey1 := servicePrefix + "instance-1"
    serviceValue1 := "192.168.1.101:8080"

    // 2. 在一个goroutine中启动服务发现监听
    go sd.DiscoverServices(servicePrefix)

    // 3. 注册一个服务实例
    fmt.Printf("Registering service: %s -> %s\n", serviceKey1, serviceValue1)
    err = sd.RegisterService(serviceKey1, serviceValue1, 10) // TTL为10秒
    if err != nil {
        log.Fatalf("Failed to register service: %v", err)
    }

    // 4. 保持运行一段时间，观察续租和监听效果
    time.Sleep(30 * time.Second)

    // 程序退出，租约到期后key会自动删除，模拟服务下线
    fmt.Println("Main function exiting. Service will be deregistered automatically when TTL expires.")
}