etcd分布式锁并不是etcd server对外提供一个性能api,而是基于etcd的各种个性(lease、watch、mvcc等)集成的一个工具。

在同一个过程外面,为了防止对共享变量产生数据竞争,通常能够通过加锁解锁的形式来防止。然而如果是多个过程,操作同一份资源,就不能用一般的锁了,这时候的“锁”须要一个能共享的介质来存储,应用它的过程能够通过一般的加锁解锁形式来防止同时操作。

思考的问题

上面咱们从一个最根本的流程剖析下一个分布式锁要思考哪些事件。

加锁

在对共享资源操作时候,首先须要加锁,在加锁时候,抢到锁的过程能够间接返回,进而操作共享资源,而没有抢到锁的过程须要期待锁的开释,对于同一个锁,同一时刻只能有一个过程来持有,这体现了锁的互斥性。

锁期间

因为是多过程状况,须要思考过程宕机的状况,如果抢到锁的过程忽然宕机,须要可能有开释锁的机制,防止前面的过程始终阻塞导致死锁。提供锁的组件也应该具备高可用性,在某个节点宕机后可能持续提供服务。

解锁

对资源的操作完结之后,须要及时开释锁,然而不能开释其余过程的锁,前面没有抢到锁的过程能够取得锁。如果抢锁的过程过多,可能会导致惊群效应,提供锁的组件应在肯定水平上防止该景象。

实现思路

etcd的几种非凡的机制都能够作为分布式锁的根底。etcd的键值对能够作为锁的本体,锁的创立与删除对应键值对的创立与删除。etcd的分布式一致性以及高可用能够保障锁的高可用性。

prefix

因为etcd反对前缀查找,能够将锁设置成“锁名称”+“惟一id”的格局,保障锁的对称性,即每个客户端只操作本人持有的锁。

lease

租约机制能够为锁做一个保活操作,在创立锁的时候绑定租约,并定期进行续约,如果取得锁期间客户端意外宕机,则持有的锁会被主动删除,防止了死锁的产生。

Revision

etcd外部保护了一个全局的Revision值,并会随着事务的递增而递增。能够用Revision值的大小来决定获取锁的先后顺序,在上锁的时候曾经决定了获取锁先后顺序,后续有客户端开释锁也不会产生惊群效应。

watch

watch机制能够用于监听锁的删除事件,不用应用忙轮询的形式查看是否开释了锁,更加高效。同时,在watch时候能够通过Revision来进行监听,只须要监听间隔本人最近而且比本人小的一个Revision就能够做到锁的实时获取。

源码剖析

在etcdv3版本的客户端库中曾经有了分布式锁的实现,让咱们看一下实现逻辑。

示例
func main() {    //初始化etcd客户端    cli, _ := clientv3.New(clientv3.Config{        Endpoints:   []string{"127.0.0.1:23790"},        DialTimeout: time.Second,    })    //创立一个session,并依据业务状况设置锁的ttl    s, _ := concurrency.NewSession(cli, concurrency.WithTTL(3))    defer s.Close()    //初始化一个锁的实例,并进行加锁解锁操作。    mu := concurrency.NewMutex(s, "mutex-linugo")    if err := mu.Lock(context.TODO()); err != nil {        log.Fatal("m lock err: ", err)    }    //do something    if err := mu.Unlock(context.TODO()); err != nil {        log.Fatal("m unlock err: ", err)    }}

在调用NewSession办法时候实际上是初始化了一个用户指定行为的租约(行为能够是指定ttl,或者复用其余的lease等),并异步进行keepalive。

type Mutex struct {    s *Session //保留的租约相干的信息    pfx   string //锁的名称,key的前缀    myKey string //锁残缺的key    myRev int64  //本人的版本号    hdr   *pb.ResponseHeader}func NewMutex(s *Session, pfx string) *Mutex {    return &Mutex{s, pfx + "/", "", -1, nil}}

NewMutex实际上创立了一个锁的数据结构,该构造能够保留一些锁的信息,入参的“mutex-linugo”只是一个key的前缀,还有后续要创立的残缺key,revision等信息。

Lock
func (m *Mutex) Lock(ctx context.Context) error {    //首先尝试获取锁    resp, err := m.tryAcquire(ctx)    if err != nil {        return err    }    ......}func (m *Mutex) tryAcquire(ctx context.Context) (*v3.TxnResponse, error) {    s := m.s    client := m.s.Client()    //残缺key是前缀名称加租约ID,因为不同过程生成的不同租约,所以锁互不雷同    m.myKey = fmt.Sprintf("%s%x", m.pfx, s.Lease())    //cmp通过比拟createRevision是否为0判断以后的key是不是第一次创立    cmp := v3.Compare(v3.CreateRevision(m.myKey), "=", 0)    //put会把key绑定上租约并存储    put := v3.OpPut(m.myKey, "", v3.WithLease(s.Lease()))    //get会获取以后key的值    get := v3.OpGet(m.myKey)    //getOwner是通过前缀来范畴查找,WithFirstCreate()筛选出以后存在的最小revision对应的值    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    }    //将该事务的revision赋值到锁的myRev字段    m.myRev = resp.Header.Revision    if !resp.Succeeded {        m.myRev = resp.Responses[0].GetResponseRange().Kvs[0].CreateRevision    }    return resp, nil}

在获取锁的时候,通过事务操作来尝试加锁。

如果以后的key是第一次创立,则将key绑定租约并存储,否则获取以后的key详细信息。getOwner通过前缀来进行查找最小revision对应的值,目标是获取以后锁的持有者(如果最小Revison的key开释锁,则该key会被删除,所以最小Revision的key就是以后锁的持有者)。

!resp.Succeeded代表key不是第一次创立,则之前执行的是get操作,获取该key创立时候的revision并赋值到锁的myRev字段。

回到主函数,目前etcd中曾经存有锁相干信息了,前面会通过比拟Revision来判断本人取得了锁还是须要期待锁,如果本人的myRev与ownerKey的Revsion雷同,阐明本人就是锁的持有者。

func (m *Mutex) Lock(ctx context.Context) error {    resp, err := m.tryAcquire(ctx)    if err != nil {        return err    }    //ownerKey就是以后持有锁的值    ownerKey := resp.Responses[1].GetResponseRange().Kvs    //如果ownerKey的长度为0或者持有者的Revision与本人的Revision雷同,阐明本人持有锁,能够间接返回,并对共享资源进行操作    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)    if werr != nil {        m.Unlock(client.Ctx())        return werr    }    //确保session没有过期    gresp, werr := client.Get(ctx, m.myKey)    if werr != nil {        m.Unlock(client.Ctx())        return werr    }    if len(gresp.Kvs) == 0 {        return ErrSessionExpired    }    m.hdr = gresp.Header    return nil}
waitDeletes

如果没有取得锁,就须要期待后面锁的开释了,这里次要用到watch机制。

func waitDeletes(ctx context.Context, client *v3.Client, pfx string, maxCreateRev int64) (*pb.ResponseHeader, error) {    //getOpts会通过两个Option函数获取小于传入的maxCreateRev的Revision的key汇合且找出汇合中最大的Revison对应的key    //次要是用于获取前一个上锁的key,进而能够watch该key的删除事件    getOpts := append(v3.WithLastCreate(), v3.WithMaxCreateRev(maxCreateRev))    for {        //get通过getOpts的动作来获取键值对        resp, err := client.Get(ctx, pfx, getOpts...)        if err != nil {            return nil, err        }        //如果长度是0,阐明key不存在,代表被删除,后面的锁曾经被开释了,能够间接返回        if len(resp.Kvs) == 0 {            return resp.Header, nil        }        lastKey := string(resp.Kvs[0].Key)        //否则通过watch监听上一个锁的删除事件        if err = waitDelete(ctx, client, lastKey, resp.Header.Revision); err != nil {            return nil, err        }    }}func waitDelete(ctx context.Context, client *v3.Client, key string, rev int64) error {    cctx, cancel := context.WithCancel(ctx)    defer cancel()    var wr v3.WatchResponse    //通过Revsion来watch key,也就是前一个锁    wch := client.Watch(cctx, key, v3.WithRev(rev))    for wr = range wch {        for _, ev := range wr.Events {             //监听Delete事件            if ev.Type == mvccpb.DELETE {                return nil            }        }    }    if err := wr.Err(); err != nil {        return err    }    if err := ctx.Err(); err != nil {        return err    }    return fmt.Errorf("lost watcher waiting for delete")}

waitDeletes失常返回后该过程会取得锁,进入操作共享资源。

UnLock

解锁操作会间接删除对应的kv,这会触发下一个锁的获取。

func (m *Mutex) Unlock(ctx context.Context) error {    client := m.s.Client()    if _, err := client.Delete(ctx, m.myKey); err != nil {        return err    }    m.myKey = "\x00"    m.myRev = -1    return nil}

小结

etcd分布式锁稳固的背地是对其自身各种个性的充分利用。本节咱们首先剖析了分布式锁满足的个性,其次列举了etcd的各种个性对分布式锁的反对状况,最初剖析了clientV3的concurrency包是怎么实现分布式锁性能的。

Reference

  • etcdV3.5.1源码 - https://github.com/etcd-io/et...
  • 拉勾教育 - 如何基于 etcd 实现分布式锁?