etcd与分布式锁的介绍

2024-04-02 19:55

短信预约 -IT技能 免费直播动态提醒

本篇内容主要讲解“etcd与分布式锁的介绍”，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习“etcd与分布式锁的介绍”吧!

1. 实现分布式锁的组件们

在分布式系统中，常用于实现分布式锁的组件有：Redis、zookeeper、etcd，下面针对各自的特性进行对比：

etcd与分布式锁的介绍

由上图可以看出三种组件各自的特点，其中对于分布式锁来说至关重要的一点是要求CP。但是，Redis集群却不支持CP，而是支持AP。虽然，官方也给出了redlock的方案，但由于需要部署多个实例(超过一半实例成功才视为成功)，部署、维护比较复杂。所以在对一致性要求很高的业务场景下(电商、银行支付)，一般选择使用zookeeper或者etcd。对比zookeeper与etcd，如果考虑性能、并发量、维护成本来看。由于etcd是用Go语言开发，直接编译为二进制可执行文件，并不依赖其他任何东西，则更具有优势。本文，则选择etcd来讨论某些观点。

2. 对于分布式锁来说AP为什么不好

在CAP理论中，由于分布式系统中多节点通信不可避免出现网络延迟、丢包等问题一定会造成网络分区，在造成网络分区的情况下，一般有两个选择：CP or AP。

① 选择AP模型实现分布式锁时，client在通过集群主节点加锁成功之后，则立刻会获取锁成功的反馈。此时，在主节点还没来得及把数据同步给从节点时发生down机的话，系统会在从节点中选出一个节点作为新的主节点，新的主节点没有老的主节点对应的锁数据，导致其他client可以在新的主节点上拿到相同的锁。这个时候，就会导致多个进程/线程/协程来操作相同的临界资源数据，从而引发数据不一致性等问题。

② 选择CP模型实现分布式锁，只有在主节点把数据同步给大于1/2的从节点之后才被视为加锁成功。此时，主节点由于某些原因down机，系统会在从节点中选取出来数据比较新的一个从节点作为新的主节点，从而避免数据丢失等问题。

所以，对于分布式锁来说，在对数据有强一致性要求的场景下，AP模型不是一个好的选择。如果可以容忍少量数据丢失，出于维护成本等因素考虑，AP模型的Redis可优先选择。

3. 分布式锁的特点以及操作

对于分布式锁来说，操作的动作包含：

获取锁
释放锁
业务处理过程中过程中，另起线程/协程进行锁的续约

etcd与分布式锁的介绍

4. 关于etcd

etcd与分布式锁的介绍

官方文档永远是最好的学习资料，官方介绍etcd如是说：

分布式系统使用etcd作为配置管理、服务发现和协调分布式工作的一致键值存储。许多组织使用etcd来实现生产系统，如容器调度器、服务发现服务和分布式数据存储。使用etcd的常见分布式模式包括leader选举、分布式锁和监视机器活动。
Distributed systems use etcd as a consistent key-value store for configuration management, service discovery, and coordinating distributed work. Many organizations use etcd to implement production systems such as container schedulers, service discovery services, and distributed data storage. Common distributed patterns using etcd include leader election, distributed locks, and monitoring machine liveness.
https://etcd.io/docs/v3.4/learning/why/

分布式锁仅是etcd可以实现众多功能中的一项，服务注册与发现在etcd中用的则会更多。

官方也对众多组件进行了对比，并整理如下：

etcd与分布式锁的介绍

通过对比可以看出各自的特点，至于具体选择哪一款，你心中可能也有了自己的答案。

5. etcd实现分布式锁的相关接口

对于分布式锁，主要用到etcd对应的添加、删除、续约接口。

// KV：键值相关操作 type KV interface {     // 存放.     Put(ctx context.Context, key, val string, opts ...OpOption) (*PutResponse, error)     // 获取.     Get(ctx context.Context, key string, opts ...OpOption) (*GetResponse, error)     // 删除.     Delete(ctx context.Context, key string, opts ...OpOption) (*DeleteResponse, error)     // 压缩rev指定版本之前的历史数据.     Compact(ctx context.Context, rev int64, opts ...CompactOption) (*CompactResponse, error)     // 通用的操作执行命令，可用于操作集合的遍历。Put/Get/Delete也是基于Do.     Do(ctx context.Context, op Op) (OpResponse, error)     // 创建一个事务，只支持If/Then/Else/Commit操作.     Txn(ctx context.Context) Txn }   // Lease：租约相关操作 type Lease interface {     // 分配一个租约.     Grant(ctx context.Context, ttl int64) (*LeaseGrantResponse, error)     // 释放一个租约.     Revoke(ctx context.Context, id LeaseID) (*LeaseRevokeResponse, error)     // 获取剩余TTL时间.     TimeToLive(ctx context.Context, id LeaseID, opts ...LeaseOption) (*LeaseTimeToLiveResponse, error)     // 获取所有租约.     Leases(ctx context.Context) (*LeaseLeasesResponse, error)     // 续约保持激活状态.     KeepAlive(ctx context.Context, id LeaseID) (<-chan *LeaseKeepAliveResponse, error)     // 仅续约激活一次.     KeepAliveOnce(ctx context.Context, id LeaseID) (*LeaseKeepAliveResponse, error)     // 关闭续约激活的功能.     Close() error }

6. etcd实现分布式锁代码示例

package main  import (     "context"     "fmt"     "go.etcd.io/etcd/clientv3"     "time" )  var conf clientv3.Config  // 锁结构体 type EtcdMutex struct {     Ttl int64//租约时间      Conf   clientv3.Config    //etcd集群配置     Key    string//etcd的key     cancel context.CancelFunc //关闭续租的func      txn     clientv3.Txn     lease   clientv3.Lease     leaseID clientv3.LeaseID }  // 初始化锁 func (em *EtcdMutex) init() error {     var err error     var ctx context.Context      client, err := clientv3.New(em.Conf)     if err != nil {         return err     }      em.txn = clientv3.NewKV(client).Txn(context.TODO())     em.lease = clientv3.NewLease(client)     leaseResp, err := em.lease.Grant(context.TODO(), em.Ttl)      if err != nil {         return err     }      ctx, em.cancel = context.WithCancel(context.TODO())     em.leaseID = leaseResp.ID     _, err = em.lease.KeepAlive(ctx, em.leaseID)      return err }  // 获取锁 func (em *EtcdMutex) Lock() error {     err := em.init()     if err != nil {         return err     }      // LOCK     em.txn.If(clientv3.Compare(clientv3.CreateRevision(em.Key), "=", 0)).         Then(clientv3.OpPut(em.Key, "", clientv3.WithLease(em.leaseID))).Else()      txnResp, err := em.txn.Commit()     if err != nil {         return err     }      // 判断txn.if条件是否成立     if !txnResp.Succeeded {         return fmt.Errorf("抢锁失败")     }      returnnil }  //释放锁 func (em *EtcdMutex) UnLock() {     // 租约自动过期，立刻过期     // cancel取消续租，而revoke则是立即过期     em.cancel()     em.lease.Revoke(context.TODO(), em.leaseID)      fmt.Println("释放了锁") }  // groutine1 func try2lock1() {     eMutex1 := &EtcdMutex{         Conf: conf,         Ttl:  10,         Key:  "lock",     }      err := eMutex1.Lock()     if err != nil {         fmt.Println("groutine1抢锁失败")         return     }     defer eMutex1.UnLock()      fmt.Println("groutine1抢锁成功")     time.Sleep(10 * time.Second) }  // groutine2 func try2lock2() {     eMutex2 := &EtcdMutex{         Conf: conf,         Ttl:  10,         Key:  "lock",     }      err := eMutex2.Lock()     if err != nil {         fmt.Println("groutine2抢锁失败")         return     }      defer eMutex2.UnLock()     fmt.Println("groutine2抢锁成功") }  // 测试代码 func EtcdRunTester() {     conf = clientv3.Config{         Endpoints:   []string{"127.0.0.1:2379"},         DialTimeout: 5 * time.Second,     }      // 启动两个协程竞争锁     go try2lock1()     go try2lock2()      time.Sleep(300 * time.Second) }