【ETCD】【实操篇（十）】基于 ETCD 实现一个简单的服务注册及发现功能

ETCD 是一个高可用的分布式键值存储系统，它广泛应用于分布式系统中作为配置中心、服务注册与发现的核心组件。通过 ETCD 实现服务注册中心，能够有效地解决服务发现与动态扩展的问题，确保各个服务能够在分布式环境中进行自动发现与连接。

在本篇文章中，我们将通过手撸一个基于 ETCD 的服务注册中心，演示如何使用 ETCD 实现服务注册与发现，并分析其技术原理。

技术原理

服务注册与发现通常需要两个关键功能：

1. 服务注册：服务向服务注册中心报告自己正在运行的信息，例如服务地址、端口、健康检查状态等。
2. 服务发现：其他服务可以通过查询服务注册中心获取到注册的服务信息，以实现动态连接。

ETCD 提供了非常适合用于服务注册与发现的特性：

• 键值存储：ETCD 是一个分布式键值存储系统，服务注册信息可以以键值对的形式存储在 ETCD 中。
• 租约机制（Lease）：ETCD 支持租约机制，使得注册的服务在超时后自动失效，非常适合用来实现服务的自动过期与清理。
• 监听机制（Watch）：ETCD 提供了监听机制，可以实时监控某个键的变化，当服务注册或注销时，监听到变化的客户端能够及时获得更新。

1. 服务注册流程

服务在启动时需要向 ETCD 注册自己的信息。这些信息通常包括：

• 服务名称（如：my-service）
• 服务实例的地址（如：http://127.0.0.1:8080）
• 服务的健康检查信息
• 服务的租约（让服务在一定时间后自动注销）

2. 服务发现流程

其他服务通过查询 ETCD 获取到已注册服务的实例信息，并与其进行通信。为了支持动态的服务发现，服务发现者可以通过监听相关的键（即服务信息的键），实时获取服务的变化。

3. 实现步骤

我们将实现一个基于 ETCD 的服务注册与发现系统，包括服务注册、心跳续约、服务注销和服务发现。

3.1 环境准备

首先，确保你已经安装并启动了 ETCD。如果你尚未安装 ETCD，可以使用以下命令进行安装和启动：

# 安装 etcd
curl -L https://github.com/etcd-io/etcd/releases/download/v3.5.0/etcd-v3.5.0-linux-amd64.tar.gz -o etcd.tar.gz
tar xzvf etcd.tar.gz
cd etcd-v3.5.0-linux-amd64

# 启动 etcd 服务
./etcd --name my-etcd-node --data-dir /tmp/etcd-data --listen-peer-urls http://127.0.0.1:2380 --listen-client-urls http://127.0.0.1:2379 --advertise-client-urls http://127.0.0.1:2379 --initial-cluster-token etcd-cluster-1 --initial-cluster my-etcd-node=http://127.0.0.1:2380 --initial-cluster-state new

3.2 使用 Go 实现服务注册与发现

我们将使用 Go 编写服务注册中心，以下是主要实现步骤：

1. 安装 Go 客户端

首先，我们需要安装 ETCD 的 Go 客户端：

go get go.etcd.io/etcd/v3

2. 创建服务注册与发现的 Go 代码

创建一个 Go 程序，完成服务注册、心跳续约、注销和发现的功能。

package main

import (
	"context"
	"fmt"
	"github.com/google/uuid" // 用于生成实例ID
	clientv3 "go.etcd.io/etcd/client/v3"
	"log"
	"time"
)

var cli *clientv3.Client

func init() {
	// 初始化etcd客户端
	var err error
	cli, err = clientv3.New(clientv3.Config{
		Endpoints:   []string{"127.0.0.1:2379"}, // 替换为你的etcd集群地址
		DialTimeout: 5 * time.Second,
	})
	if err != nil {
		log.Fatal(err)
	}
}

// 服务注册
func registerService(serviceName, serviceAddress string) (clientv3.LeaseID, error) {
	// 生成一个唯一的实例ID
	instanceID := uuid.New().String()

	// 创建租约
	grantResp, err := cli.Grant(context.Background(), 10)
	if err != nil {
		return 0, err
	}

	// 将服务信息注册到 ETCD，使用 serviceName + instanceID 作为唯一的注册键
	key := fmt.Sprintf("/services/%s/%s", serviceName, instanceID)
	_, err = cli.Put(context.Background(), key, serviceAddress, clientv3.WithLease(grantResp.ID))
	if err != nil {
		return 0, err
	}
	cli.KeepAlive(context.Background(), grantResp.ID)

	// 返回租约 ID，后续可以用来续约
	return grantResp.ID, nil
}

// 服务注销
func deregisterService(serviceName, instanceID string) error {
	key := fmt.Sprintf("/services/%s/%s", serviceName, instanceID)
	_, err := cli.Delete(context.Background(), key)
	return err
}

// 服务发现
func discoverService(serviceName string) ([]string, error) {
	// 查询所有与 serviceName 相关的服务实例
	keyPrefix := fmt.Sprintf("/services/%s", serviceName)
	resp, err := cli.Get(context.Background(), keyPrefix, clientv3.WithPrefix())
	if err != nil {
		return nil, err
	}
	if len(resp.Kvs) == 0 {
		return nil, fmt.Errorf("没有找到该服务的实例")
	}

	// 提取所有服务实例地址
	var serviceAddresses []string
	for _, kv := range resp.Kvs {
		serviceAddresses = append(serviceAddresses, string(kv.Value))
	}
	return serviceAddresses, nil
}

func main() {
	// 连接到 ETCD 集群
	var err error
	cli, err = clientv3.New(clientv3.Config{
		Endpoints:   []string{"http://127.0.0.1:2379"},
		DialTimeout: 5 * time.Second,
	})
	if err != nil {
		log.Fatal(err)
	}
	defer cli.Close()

	// 注册多个服务实例
	_, err = registerService("my-service", "http://127.0.0.1:8081")
	if err != nil {
		log.Fatalf("服务注册失败: %v", err)
	}
	log.Println("服务实例 1 注册成功，地址：http://127.0.0.1:8081")

	_, err = registerService("my-service", "http://127.0.0.1:8082")
	if err != nil {
		log.Fatalf("服务注册失败: %v", err)
	}
	log.Println("服务实例 2 注册成功，地址：http://127.0.0.1:8082")

	// 服务发现
	serviceAddresses, err := discoverService("my-service")
	if err != nil {
		log.Printf("服务发现失败: %v", err)
	} else {
		log.Printf("发现服务实例：%v", serviceAddresses)
	}

	// 模拟服务注销
	time.Sleep(20 * time.Second)
	err = deregisterService("my-service", "instance-1") // 示例：注销特定实例
	if err != nil {
		log.Fatalf("服务注销失败: %v", err)
	}
	log.Println("服务实例 1 已注销")

	// 服务发现
	serviceAddresses, err = discoverService("my-service")
	if err != nil {
		log.Printf("服务发现失败: %v", err)
	} else {
		log.Printf("发现服务实例：%v", serviceAddresses)
	}
}

3.3 代码分析

1. 注册服务：服务通过 registerService 函数向 ETCD 注册自己的信息，使用了 ETCD 的租约机制确保服务在指定时间后自动注销。如果需要续约，服务可以通过 keepAliveService 进行定期续约。

2. 服务注销：如果服务关闭或退出，我们可以使用 deregisterService 函数显式删除注册信息。

3. 服务发现：服务发现者通过 discoverService 函数从 ETCD 获取服务注册信息，动态地获取服务地址。

4. 租约与心跳：每个服务注册都有一个租约，租约会在一定时间后过期，服务可以定期续约来延长租期，避免服务被自动清除。

3.4 运行服务注册中心

运行上述 Go 程序后，你可以看到服务会注册到 ETCD 中，服务会持续进行续约，直到程序终止。其他服务可以通过相同的 discoverService 方法进行服务发现。

4. 总结

通过基于 ETCD 实现的服务注册中心，服务能够方便地注册到 ETCD 中，利用 ETCD 的租约机制实现自动注销与过期。同时，借助 ETCD 提供的监听机制，其他服务可以实时发现注册信息，动态连接到服务。这种方式简单而高效，适用于分布式系统中的服务注册与发现。

在实际应用中，可以通过增强健康检查、增加负载均衡、支持多租户等方式对该服务注册中心进行扩展，使其满足更多的实际需求。