分布式锁具备特点
-
互斥性：在同一时刻，只有一个客户端能持有锁
-
安全性：避免死锁，如果某个客户端获得锁之后处理时间超过最大约定时间，或者持锁期间发生了故障导致无法主动释放锁，其持有的锁也能够被其他机制正确释放，并保证后续其它客户端也能加锁，整个处理流程继续正常执行
-
可用性：也被称作容错性，分布式锁需要有高可用能力，避免单点故障，当提供锁的服务节点故障（宕机）时不影响服务运行，这里有两种模式：一种是分布式锁服务自身具备集群模式，遇到故障能自动切换恢复工作；另一种是客户端向多个独立的锁服务发起请求，当某个锁服务故障时仍然可以从其他锁服务读取到锁信息(Redlock)
-
可重入性：对同一个锁，加锁和解锁必须是同一个线程，即不能把其他线程持有的锁给释放了
-
高效灵活：加锁、解锁的速度要快；支持阻塞和非阻塞；支持公平锁和非公平锁
Redis实现分布式锁的缺点
-
客户端长时间阻塞导致锁失效问题
客户端A获取锁后在处理业务时长时间阻塞，导致锁过期释放。当阻塞恢复时，就会出现多客户端同时持有锁的情况。
redis的解决方案：
– 延长锁的过期时间。
– java redission 提供了看门狗机制，在业务处理完之前不断给锁续期。 -
单点实例安全性问题
为了保证分布式锁的高可用，需要部署redis的主从节点，在数据同步之前发生主从切换，可能就会丢失原先master上的锁信息，导致同一时间两个客户端同时持有锁。
redis的解决方案：
参考官方文档redlock
ETCD实现分布式锁的思路
prefix
etcd支持前缀查找，所以可以用一个前缀表示锁资源，前缀 + 唯一id的方式表示锁资源的持有者。
lease机制
租约机制可以保证锁的活性，持有锁的客户端宕机，key自动过期，避免宕机。etcd客户端提供的lease续租机制解决客户端长时间阻塞导致锁失效问题。
watch机制
redis采用忙轮询的方式来获取锁，etcd可以使用watch机制监听锁的删除事件，更加高效。
实现策略
etcd实现分布式锁的方案有很多种，可以通过判断是否存在一个固定的key来实现分布式锁，但是这种实现策略有很大的问题。当多客户端同时获取锁时，只有一个成功获得，其余多个客户端监听key的删除事件，一旦锁被释放，多个客户端同时收到锁删除事件(无论尝试加锁的顺序)进行加锁，这就是 “惊群问题” ，所以etcd官网提供了另外一种实现策略。
不再将一个固定的key当作锁资源，而是将一个前缀当作锁资源，每一个客户端尝试加锁的时候都会以该前缀创建一个key，并且监听前一个创建该前缀key的版本号。
具体的实现方式会在源码解析时讲解。
etcd的强一致性
etcd是基于raft实现的，对外提供的是强一致性的kv存储，不会存在类似于redis主从切换导致的不一致问题。
etcd客户端concurrency包提供的分布式锁
锁的使用
func NewLock() sync.Locker {
//创建客户端
cli, err := clientv3.New(clientv3.Config{Endpoints: ip1,ip2...})
if err != nil {
log.Fatal(err)
}
//授权租约
resp, err := cli.Grant(context.TODO(), 5)
if err != nil {
log.Fatal(err)
}
//创建会话
//会话会创建一个租约，并在客户端生存期内保证租约的活性
session, err := concurrency.NewSession(cli, concurrency.WithLease(resp.ID))
if err != nil {
log.Fatal(err)
}
//利用会话，指定一个前缀创建锁
return concurrency.NewLocker(session, "/myLock/")
}
通过以上方式就可以创建一个分布式锁，通过锁的Lock()和UnLock()方法加锁解锁。
源码解析
type Mutex struct {
s *Session //会话
pfx string //锁名称，key的前缀
myKey string //key
myRev int64 //创建key的版本号
hdr *pb.ResponseHeader
}
加锁
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，prefix + leaseId
m.myKey = fmt.Sprintf("%s%x", m.pfx, s.Lease())
// 比较操作，判断当前key的创建版本号是否为0,版本号为0表示key还未创建
cmp := v3.Compare(v3.CreateRevision(m.myKey), "=", 0)
// 创建key操作(将当前key存储到etcd)
p**加锁**ut := v3.OpPut(m.myKey, "", v3.WithLease(s.Lease()))
// 获取当前key的版本号操作
get := v3.OpGet(m.myKey)
// 获取第一个创建该前缀key的版本号(不会获取到已经删除的key的版本号)，这个key就是当前持有锁的key
getOwner := v3.OpGet(m.pfx, v3.WithFirstCreate()...)
//通过一个事务执行操作
//若当前key不存在，创建key，并获取持有锁的key的版本号
//若当前key存在，获取key的版本号，并获取持有锁的key
resp, err := client.Txn(ctx).If(cmp).Then(put, getOwner).Else(get, getOwner).Commit()
if err != nil {
return nil, err
}
//将当前key的把版本号赋值给myRev字段
m.myRev = resp.Header.Revision
if !resp.Succeeded {
m.myRev = resp.Responses[0].GetResponseRange().Kvs[0].CreateRevision
}
return resp, nil
}
func (m *Mutex) Lock(ctx context.Context) error {
resp, err := m.tryAcquire(ctx)
if err != nil {
return err
}
// 将当前持有锁的key赋值给ownerKey
ownerKey := resp.Responses[1].GetResponseRange().Kvs
//ownerKey不存在，或者版本号等于自己创建key的版本号，表示当前key正持有锁，可直接返回
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)
if werr != nil {
m.Unlock(client.Ctx())
return werr
}
if len(gresp.Kvs) == 0 { // is the session key lost?
return ErrSessionExpired
}
m.hdr = gresp.Header
return nil
}
//等待锁的释放
func waitDeletes(ctx context.Context, client *v3.Client, pfx string, maxCreateRev int64) (*pb.ResponseHeader, error) {
//获取最新的该前缀的key的操作，WithMaxCreateRev(maxCreateRev)对返回值进行了限制，返回值版本号必须是小于等于maxCreateRev的，通过这个操作就可以获取仅小于自己key版本号的key
getOpts := append(v3.WithLastCreate(), v3.WithMaxCreateRev(maxCreateRev))
for {
resp, err := client.Get(ctx, pfx, getOpts...)
if err != nil {
return nil, err
}
//不存在大于自己版本号的key，可以获取锁了
if len(resp.Kvs) == 0 {
return resp.Header, nil
}
lastKey := string(resp.Kvs[0].Key)
//等待 lastKey 被删除
if err = waitDelete(ctx, client, lastKey, resp.Header.Revision); err != nil {
return nil, err
}
}
}
//通过watch机制监听指定version 的key的删除。
func waitDelete(ctx context.Context, client *v3.Client, key string, rev int64) error {
cctx, cancel := context.WithCancel(ctx)
defer cancel()
var wr v3.WatchResponse
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")
}
释放锁
func (m *Mutex) Unlock(ctx context.Context) error {
client := m.s.Client()
//直接删除key
if _, err := client.Delete(ctx, m.myKey); err != nil {
return err
}
m.myKey = "\x00"
m.myRev = -1
return nil
}
总结
基于etcd实现的分布式锁基本上使用到了etcd的全部性质，并且保证了分布式锁的互斥性，安全性和可用性。官方实现的分布式锁并不支持可重入性，但是要实现可重入性锁也很简单，对这个锁在封装一层，并增加一个计数器。
参考资料
极客时间- etcd实战课
etcd分布式锁的实现原理
![[Linux] 动态 / 静态库的生成与使用](https://img-blog.csdnimg.cn/59a5f24d2c13419fa62422456f5b64b3.png)
