· code
Golang分布式应用之etcd
etcd 是一个可靠的分布式 KV 存储数据库，由 CoreOS 开源。Kuberentes 使用 etcd 作为其存储引擎，随着云原生的火热，etcd 也逐渐广泛应用起来。
etcd 除了作为普通的 KV 存储、配置存储，还可以用在以下分布式场景中：
- 服务发现
- 分布式锁
- 选主
- 分布式队列
- 分布式系统协调
- 负载均衡
本文结合 Golang 来编写对应的中间件，所有代码见https://github.com/qingwave/gocorex
服务发现
在分布式系统中，如何能找到所需要访问的服务即服务发现。服务较少时可以直接访问其 IP，但随着业务规模的扩大，维护其地址越来越复杂，如果服务频繁的扩缩容，必须能够实时感应服务的断点变化。 通常有多种方式可以解决
- 系统级别，如 LVS、DNS、Kubernetes 中的 Service、Istio 等
- 微服务注册中心，如 Spring Cloud 中的 Enruka，Dubbo 等
- 借助分布式协调系统 etcd、ZK、Consul 等
服务发现提供的功能包括：
- 服务注册、注销
- 服务宕机或异常时，自动注销
- 感知服务端点变化
借助 etcd 实现服务发现
- 可以通过将端点写同一个目录(相同前缀，如/services/job/endpoint1, /services/job/endpoint2)，并通过 Lease 设置一个过期时间，不断刷新 Lease，如果服务宕机，Lease 过期对应端点会自动删除
- 通过 Watch API 可以监听端点变化
主要代码如下
func New(config EtcdDiscoveryConfig) (*EtcdDiscovery, error) {
// 创建session，session会自动续约
session, err := concurrency.NewSession(config.Client, concurrency.WithTTL(config.TTLSeconds))
if err != nil {
return nil, err
}
config.Prefix = strings.TrimSuffix(config.Prefix, "/") + "/"
return &EtcdDiscovery{
EtcdDiscoveryConfig: config,
session: session,
myKey: config.Prefix + config.Key,
services: make(map[string]string),
}, nil
}
func (d *EtcdDiscovery) Register(ctx context.Context) error {
lease := d.session.Lease()
// 注册服务
_, err := d.Client.Put(ctx, d.myKey, d.Val, clientv3.WithLease(lease))
return err
}
func (d *EtcdDiscovery) UnRegister(ctx context.Context) error {
// 注销服务
_, err := d.Client.Delete(ctx, d.myKey)
return err
}
// 监听端点变化
func (d *EtcdDiscovery) Watch(ctx context.Context) error {
// context用来停止监听
d.watchContext, d.watchCancel = context.WithCancel(ctx)
// 首先获取所有端点
resp, err := d.Client.Get(d.watchContext, d.Prefix, clientv3.WithPrefix())
services := make(map[string]string)
for _, kv := range resp.Kvs {
services[string(kv.Key)] = string(kv.Value)
}
d.setServices(services)
// 回调点，用户可自定义
if d.Callbacks.OnStartedDiscovering != nil {
d.Callbacks.OnStartedDiscovering(d.ListServices())
}
defer func() {
if d.Callbacks.OnStoppedDiscovering != nil {
d.Callbacks.OnStoppedDiscovering()
}
}()
defer d.watchCancel()
// 监听目录，通过WithPrefix可以添加子目录变化
ch := d.Client.Watch(d.watchContext, d.Prefix, clientv3.WithPrefix())
for {
select {
case <-d.watchContext.Done():
return nil
case wr, ok := <-ch:
if !ok {
return fmt.Errorf("watch closed")
}
if wr.Err() != nil {
return wr.Err()
}
// 将添加事件同步到本地端点列表
for _, ev := range wr.Events {
key, val := string(ev.Kv.Key), string(ev.Kv.Value)
switch ev.Type {
case mvccpb.PUT:
d.addService(key, val)
case mvccpb.DELETE:
d.delService(key)
}
if d.Callbacks.OnServiceChanged != nil {
event := DiscoveryEvent{Type: mvccpb.Event_EventType_name[int32(ev.Type)], Service: d.serviceFromKv(key, val)}
d.Callbacks.OnServiceChanged(d.ListServices(), event)
}
}
}
}
}
主要实现逻辑如下：
- 创建 Session， Session 中 Lease 会自动续约
- 服务注册时，在目录下创建对应的子目录，并附带 Lease
- 通过 Watch 接口监听目录变化，同步到本地
简单测试下，通过 worker 模拟不同的端点
func main() {
client, err := clientv3.New(clientv3.Config{
Endpoints: []string{"localhost:2379"},
DialTimeout: 3 * time.Second,
})
if err != nil {
log.Fatalf("failed to create etcd lock: %v", err)
}
defer client.Close()
worker := func(i int, run bool) {
id := fmt.Sprintf("worker-%d", i)
val := fmt.Sprintf("10.0.0.%d", i)
sd, err := etcdiscovery.New(etcdiscovery.EtcdDiscoveryConfig{
Client: client,
Prefix: "/services",
Key: id,
Val: val,
TTLSeconds: 2,
Callbacks: etcdiscovery.DiscoveryCallbacks{
OnStartedDiscovering: func(services []etcdiscovery.Service) {
log.Printf("[%s], onstarted, services: %v", id, services)
},
OnStoppedDiscovering: func() {
log.Printf("[%s], onstoped", id)
},
OnServiceChanged: func(services []etcdiscovery.Service, event etcdiscovery.DiscoveryEvent) {
log.Printf("[%s], onchanged, services: %v, event: %v", id, services, event)
},
},
})
if err != nil {
log.Fatalf("failed to create service etcdiscovery: %v", err)
}
defer sd.Close()
if !run {
if sd.UnRegister(context.Background()); err != nil {
log.Fatalf("failed to unregister service [%s]: %v", id, err)
}
return
}
if err := sd.Register(context.Background()); err != nil {
log.Fatalf("failed to register service [%s]: %v", id, err)
}
if err := sd.Watch(context.Background()); err != nil {
log.Fatalf("failed to watch service: %v", err)
}
}
wg := group.NewGroup()
for i := 0; i < 3; i++ {
id := i
wg.Go(func() { worker(id, true) })
}
go func() {
time.Sleep(2 * time.Second)
worker(3, true)
}()
// unregister
go func() {
time.Sleep(4 * time.Second)
worker(2, false)
}()
wg.Wait()
}
通过结果可以看到，服务可以正常的注册注销，并能实时监听端点变化
2022/08/08 08:44:02 [worker-1], onstarted, services: [{/services/worker-1 worker-1 10.0.0.1} {/services/worker-2 worker-2 10.0.0.2} {/services/worker-0 worker-0 10.0.0.0}]
2022/08/08 08:44:02 [worker-2], onstarted, services: [{/services/worker-0 worker-0 10.0.0.0} {/services/worker-1 worker-1 10.0.0.1} {/services/worker-2 worker-2 10.0.0.2}]
2022/08/08 08:44:02 [worker-0], onstarted, services: [{/services/worker-0 worker-0 10.0.0.0} {/services/worker-1 worker-1 10.0.0.1} {/services/worker-2 worker-2 10.0.0.2}]
2022/08/08 08:44:04 [worker-2], onchanged, services: [{/services/worker-0 worker-0 10.0.0.0} {/services/worker-1 worker-1 10.0.0.1} {/services/worker-2 worker-2 10.0.0.2} {/services/worker-3 worker-3 10.0.0.3}], event: {PUT {/services/worker-3 worker-3 10.0.0.3}}
2022/08/08 08:44:04 [worker-1], onchanged, services: [{/services/worker-0 worker-0 10.0.0.0} {/services/worker-1 worker-1 10.0.0.1} {/services/worker-2 worker-2 10.0.0.2} {/services/worker-3 worker-3 10.0.0.3}], event: {PUT {/services/worker-3 worker-3 10.0.0.3}}
2022/08/08 08:44:04 [worker-0], onchanged, services: [{/services/worker-3 worker-3 10.0.0.3} {/services/worker-0 worker-0 10.0.0.0} {/services/worker-1 worker-1 10.0.0.1} {/services/worker-2 worker-2 10.0.0.2}], event: {PUT {/services/worker-3 worker-3 10.0.0.3}}
2022/08/08 08:44:04 [worker-3], onstarted, services: [{/services/worker-0 worker-0 10.0.0.0} {/services/worker-1 worker-1 10.0.0.1} {/services/worker-2 worker-2 10.0.0.2} {/services/worker-3 worker-3 10.0.0.3}]
2022/08/08 08:44:06 [worker-1], onchanged, services: [{/services/worker-0 worker-0 10.0.0.0} {/services/worker-1 worker-1 10.0.0.1} {/services/worker-3 worker-3 10.0.0.3}], event: {DELETE {/services/worker-2 worker-2 }}
2022/08/08 08:44:06 [worker-3], onchanged, services: [{/services/worker-3 worker-3 10.0.0.3} {/services/worker-0 worker-0 10.0.0.0} {/services/worker-1 worker-1 10.0.0.1}], event: {DELETE {/services/worker-2 worker-2 }}
2022/08/08 08:44:06 [worker-0], onchanged, services: [{/services/worker-0 worker-0 10.0.0.0} {/services/worker-1 worker-1 10.0.0.1} {/services/worker-3 worker-3 10.0.0.3}], event: {DELETE {/services/worker-2 worker-2 }}
2022/08/08 08:44:06 [worker-2], onchanged, services: [{/services/worker-0 worker-0 10.0.0.0} {/services/worker-1 worker-1 10.0.0.1} {/services/worker-3 worker-3 10.0.0.3}], event: {DELETE {/services/worker-2 worker-2 }}
分布式锁
在 ECTD 官方库go.etcd.io/etcd/client/v3/concurrency中，已经支持分布式锁。
主要原理与之前通过Redis实现的分布式锁类似，如果目录创建成功则加锁成功，解锁直接删除即可。
etcd 锁的使用
// 创建session并不断刷新
session, err := concurrency.NewSession(client, concurrency.WithTTL(2*time.Second))
if err != nil {
return nil, err
}
mutex := concurrency.NewMutex(session, config.Prefix)
mutex.Lock()
defer mutext.UnLock()
do()....
加锁的核心逻辑如下
func (m *Mutex) tryAcquire(ctx context.Context) (*v3.TxnResponse, error) {
s := m.s
client := m.s.Client()
m.myKey = fmt.Sprintf("%s%x", m.pfx, s.Lease())
cmp := v3.Compare(v3.CreateRevision(m.myKey), "=", 0)
// put self in lock waiters via myKey; oldest waiter holds lock
put := v3.OpPut(m.myKey, "", v3.WithLease(s.Lease()))
// reuse key in case this session already holds the lock
get := v3.OpGet(m.myKey)
// fetch current holder to complete uncontended path with only one RPC
getOwner := v3.OpGet(m.pfx, v3.WithFirstCreate()...)
resp, err := client.Txn(ctx).If(cmp).Then(put, getOwner).Else(get, getOwner).Commit()
if err != nil {
return nil, err
}
m.myRev = resp.Header.Revision
if !resp.Succeeded {
m.myRev = resp.Responses[0].GetResponseRange().Kvs[0].CreateRevision
}
return resp, nil
}
tryAcquire 通过事务来执行加锁逻辑:
- 判断当前 Key 是否为空，即代码中 Revision 为 0
- 如果为空，使用 Put 设置并附加 Lease
- 如果不为空，获取当前锁的所有者，即最先加锁的对象，避免惊群效应
func (m *Mutex) Lock(ctx context.Context) error {
resp, err := m.tryAcquire(ctx)
if err != nil {
return err
}
// if no key on prefix / the minimum rev is key, already hold the lock
ownerKey := resp.Responses[1].GetResponseRange().Kvs
if len(ownerKey) == 0 || ownerKey[0].CreateRevision == m.myRev {
m.hdr = resp.Header
return nil
}
client := m.s.Client()
_, werr := waitDeletes(ctx, client, m.pfx, m.myRev-1)
// release lock key if wait failed
if werr != nil {
m.Unlock(client.Ctx())
return werr
}
// make sure the session is not expired, and the owner key still exists.
gresp, werr := client.Get(ctx, m.myKey)
return nil
}
Lock 方法会一直阻塞，直到获取锁返回执行出错:
- 调用 tryAcquire
- 如果已经加锁成功，或者已经加过锁（可重入），则直接返回
- 调用 waitDeletes 方法，等待所有小于当前 Revsion 的 Key 删除
分布式选主
对于有状态的服务，为了提供其服务水平 SLA 减少宕机时间，通过会有多个副本，当主节点宕机时，副本节点可以快速切换。
通过 etcd 可以实现选主服务，与分布式比较类似
- 选主成功，不断上报心跳
- 通过 Watch 接口，当节点失效时，去竞争主(类似加锁过程)
在 ECTD 官方库go.etcd.io/etcd/client/v3/concurrency中，已经支持了分布式选主。
选主核心逻辑如下
func (e *Election) Campaign(ctx context.Context, val string) error {
s := e.session
client := e.session.Client()
k := fmt.Sprintf("%s%x", e.keyPrefix, s.Lease())
txn := client.Txn(ctx).If(v3.Compare(v3.CreateRevision(k), "=", 0))
txn = txn.Then(v3.OpPut(k, val, v3.WithLease(s.Lease())))
txn = txn.Else(v3.OpGet(k))
resp, err := txn.Commit()
if err != nil {
return err
}
e.leaderKey, e.leaderRev, e.leaderSession = k, resp.Header.Revision, s
if !resp.Succeeded {
kv := resp.Responses[0].GetResponseRange().Kvs[0]
e.leaderRev = kv.CreateRevision
if string(kv.Value) != val {
if err = e.Proclaim(ctx, val); err != nil {
e.Resign(ctx)
return err
}
}
}
_, err = waitDeletes(ctx, client, e.keyPrefix, e.leaderRev-1)
if err != nil {
// clean up in case of context cancel
select {
case <-ctx.Done():
e.Resign(client.Ctx())
default:
e.leaderSession = nil
}
return err
}
e.hdr = resp.Header
return nil
}
以上逻辑与 ECTD 锁中的实现非常相似
- 开启事务，首先判断当前服务 Key 是否存在
- 不存在，通过 Put 设置对应值
- 存在获得当前目录最小 Revision 的值，即当前主节点
- 通过 waitDeletes，直到当前进程的 Revision
简单封装下，支持回调，参考了 Kubernetes 的选主实现
func New(config LeaderElectionConfig) (*EctdLeaderElection, error) {
session, err := concurrency.NewSession(config.Client, concurrency.WithTTL(config.LeaseSeconds))
if err != nil {
return nil, err
}
election := concurrency.NewElection(session, config.Prefix)
return &EctdLeaderElection{
LeaderElectionConfig: config,
session: session,
election: election,
}, nil
}
// 运行选主
func (le *EctdLeaderElection) Run(ctx context.Context) error {
defer func() {
le.Callbacks.OnStoppedLeading()
}()
ctx, cancel := context.WithCancel(ctx)
defer cancel()
// 添加选主变化
go le.observe(ctx)
// 开始选主
if err := le.election.Campaign(ctx, le.Identity); err != nil {
return err
}
// 选主完成，运行OnStarted，运行结束则退出选主
le.Callbacks.OnStartedLeading(ctx)
return nil
}
// 监听Key变化，执行回调
func (le *EctdLeaderElection) observe(ctx context.Context) {
if le.Callbacks.OnNewLeader == nil {
return
}
ch := le.election.Observe(ctx)
for {
select {
case <-ctx.Done():
return
case resp, ok := <-ch:
if !ok {
return
}
if len(resp.Kvs) == 0 {
continue
}
leader := string(resp.Kvs[0].Value)
if leader != le.Identity {
go le.Callbacks.OnNewLeader(leader)
}
}
}
}
func (le *EctdLeaderElection) Close() error {
return le.session.Close()
}
测试选主服务
func main() {
client, err := clientv3.New(clientv3.Config{
Endpoints: []string{"localhost:2379"},
DialTimeout: 3 * time.Second,
})
if err != nil {
log.Fatalf("failed to create etcd lock: %v", err)
}
defer client.Close()
prefix := "/worker/election"
worker := func(i int) {
id := fmt.Sprintf("worker-%d", i)
le, err := leaderelection.New(leaderelection.LeaderElectionConfig{
Client: client,
LeaseSeconds: 15,
Prefix: prefix,
Identity: id,
Callbacks: leaderelection.LeaderCallbacks{
OnStartedLeading: func(ctx context.Context) {
log.Printf("OnStarted[%s]: acquire new leader", id)
time.Sleep(3 * time.Second)
log.Printf("OnStarted[%s]: worker done", id)
},
OnStoppedLeading: func() {
log.Printf("OnStopped[%s]: exit", id)
},
OnNewLeader: func(identity string) {
log.Printf("OnNewLeader[%s]: new leader %s", id, identity)
},
},
})
if err != nil {
log.Fatalf("failed to create leader election: %v", err)
}
defer le.Close()
le.Run(context.Background())
}
wg := sync.WaitGroup{}
for i := 1; i <= 3; i++ {
wg.Add(1)
id := i
go func() {
defer wg.Done()
worker(id)
}()
}
wg.Wait()
}
运行结果
2022/08/08 09:33:32 OnNewLeader[worker-2]: new leader worker-3
2022/08/08 09:33:32 OnNewLeader[worker-1]: new leader worker-3
2022/08/08 09:33:32 OnStarted[worker-3]: acquire new leader
2022/08/08 09:34:02 OnStarted[worker-3]: worker done
2022/08/08 09:34:02 OnStopped[worker-3]: exit
2022/08/08 09:34:02 OnStarted[worker-2]: acquire new leader
2022/08/08 09:34:02 OnNewLeader[worker-1]: new leader worker-2
2022/08/08 09:34:32 OnStarted[worker-2]: worker done
2022/08/08 09:34:32 OnStopped[worker-2]: exit
2022/08/08 09:34:32 OnStarted[worker-1]: acquire new leader
2022/08/08 09:35:02 OnStarted[worker-1]: worker done
2022/08/08 09:35:02 OnStopped[worker-1]: exit
发布订阅
借助 etcd 的前缀查找、Watch 的功能，可以实现发布订阅功能，主要逻辑如下
// 发布时，直接通过Put将对象设置在对应Topic路径下，并可以设置Lease，自动删除过时消息
func (ps *EtcdPubSub) Publish(ctx context.Context, topic string, msg Msg) error {
le, err := ps.Client.Lease.Grant(ctx, int64(ps.TTLSeconds))
if err != nil {
return err
}
_, err = ps.Client.Put(ctx, ps.Prefix+topic+"/"+msg.Name, msg.Val, clientv3.WithLease(le.ID))
return err
}
// 订阅时，通过Watch来监听Topic是否有Put事件，这里忽略Delete事件
// Revision为0时，从当前时间点开始监听
// Revision为1时，监听Topic创建后的所有事件
func (ps *EtcdPubSub) SubscribeFromRev(ctx context.Context, topic string, rev int64) (<-chan Msg, error) {
wch := ps.Client.Watch(ctx, ps.Prefix+topic, clientv3.WithPrefix(), clientv3.WithFilterDelete(), clientv3.WithRev(rev))
msg := make(chan Msg)
go func() {
defer close(msg)
for {
wc, ok := <-wch
if !ok {
return
}
for _, ev := range wc.Events {
if ev.Type != mvccpb.PUT {
break
}
name := strings.TrimPrefix(string(ev.Kv.Key), ps.Prefix+topic+"/")
msg <- Msg{Name: name, Val: string(ev.Kv.Value)}
}
}
}()
return msg, nil
}
发布时，直接通过 PUT 操作在 Topic 路径下设置消息； 订阅时，通过 Watch 来捕获消息，通过 Revision 来配置不同的监听行为
- Revision 为 0 时，从当前时间点开始监听
- Revision 为 1 时，监听 Topic 创建后的所有事件
总结
本文主要结合 Golang 总结了 etcd 中服务发现、分布式锁、选主等实现方式，另外 etcd 还可以应用在发布订阅、负载均衡等方面。
本文所有代码见https://github.com/qingwave/gocorex，欢迎批评指正。
Explore more in https://qingwave.github.io