etcd Lease：etcd 如何实现租约？

$ etcdctl lease grant 1000

lease 694d77aa9e38260f granted with ttl(1000s)

$ etcdctl lease timetolive 694d77aa9e38260f

lease 694d77aa9e38260f granted with ttl(1000s), remaining(983s)

$ etcdctl put foo bar --lease 694d77aa9e38260f
OK

# 等待过期，再次查看租约信息
$ etcdctl lease timetolive 694d77aa9e38260f

lease 694d77aa9e38260f already expired

// 位于 lease/lessor.go:82
type Lessor interface {
    //...省略部分
    // 将 lessor 设置为 Primary，这个与 raft 会出现网络分区有关
    Promote(extend time.Duration)

 // Grant 创建了一个在指定时间过期的 Lease 对象
 Grant(id LeaseID, ttl int64) (*Lease, error)
 // Revoke 撤销指定 LeaseID，绑定到其上的键值对将会被移除，如果该 LeaseID 对应的 Lease 不存在，则会返回错误
 Revoke(id LeaseID) error

 // Attach 绑定给定的 LeaseItem 到 LeaseID，如果该租约不存在，将会返回错误
 Attach(id LeaseID, items []LeaseItem) error

 // GetLease 返回 LeaseItem 对应的 LeaseID
 GetLease(item LeaseItem) LeaseID

 // Detach 将 LeaseItem 从给定的 LeaseID 解绑。如果租约不存在，则会返回错误
 Detach(id LeaseID, items []LeaseItem) error

 // Renew 刷新指定 LeaseID，结果将会返回刷新后的 TTL
 Renew(id LeaseID) (int64, error)

 // Lookup 查找指定的 LeaseID，返回对应的 Lease
 Lookup(id LeaseID) *Lease

 // Leases 方法列出所有的 Leases
 Leases() []*Lease

 // ExpiredLeasesC 用于返回接收过期 Lease 的 channel
 ExpiredLeasesC() <-chan []*Lease

}

// 位于 lease/lessor.go:800
type Lease struct {
 ID           LeaseID
 ttl          int64 // 存活时间，单位秒
 remainingttl int64 // 剩余的存活时间，如果为 0，则被认为是未设置，这种情况下该值与 TTL 相等
 // expiry 的并发锁
 expiryMu sync.RWMutex
 // expiry 是 Lease 过期的时间，当expiry.IsZero() 为 true 时，则没有过期时间
 expiry time.Time

 // ItemSet 并发锁
 mu      sync.RWMutex
 itemSet map[LeaseItem]struct{}
 revokec chan struct{}
}

type lessor struct {
 mu sync.RWMutex

 demotec chan struct{}

 leaseMap             map[LeaseID]*Lease
 leaseExpiredNotifier *LeaseExpiredNotifier
 leaseCheckpointHeap  LeaseQueue
 itemMap              map[LeaseItem]LeaseID

 // 当 Lease 过期，lessor 将会通过 RangeDeleter 删除相应范围内的 keys
 rd RangeDeleter

 cp Checkpointer

 // backend 目前只会保存 LeaseID 和 expiry。LeaseItem 通过遍历 kv 中的所有键来恢复
 b backend.Backend

 // minLeasettl 是最小的 TTL 时间
 minLeasettl int64

 expiredC chan []*Lease
 // stopC 用来表示 lessor 应该被停止的 channel
 stopC chan struct{}
 // doneC 用来表示 lessor 已经停止的 channel
 doneC chan struct{}

 lg *zap.Logger

 checkpointInterval time.Duration

 expiredLeaseRetryInterval time.Duration
}

// 位于 lease/lessor.go:258
func (le *lessor) Grant(id LeaseID, ttl int64) (*Lease, error) {
  // TTL 不能大于 MaxLeasettl
 if ttl > MaxLeasettl {
  return nil, ErrLeasettlTooLarge
 }

 // 构建 Lease 对象
 l := &Lease{
  ID:      id,
  ttl:     ttl,
  itemSet: make(map[LeaseItem]struct{}),
  revokec: make(chan struct{}),
 }

 le.mu.Lock()
 defer le.mu.Unlock()
    // 查找内存 LeaseMap 中是否有 LeaseID 对应的 Lease
 if _, ok := le.leaseMap[id]; ok {
  return nil, ErrLeaseExists
 }

 if l.ttl < le.minLeasettl {
  l.ttl = le.minLeasettl
 }

 if le.isPrimary() {
  l.refresh(0)
 } else {
  l.forever()
 }
    // 将 l 存放到 LeaseMap 和 LeaseExpiredNotifier 
 le.leaseMap[id] = l
 item := &LeaseWithTime{id: l.ID, time: l.expiry.UnixNano()}
 le.leaseExpiredNotifier.RegisterOrUpdate(item)
 l.persistTo(le.b)

 leaseTotalttls.Observe(float64(l.ttl))
 leaseGranted.Inc()

 if le.isPrimary() {
  le.scheduleCheckpointIfNeeded(l)
 }

 return l, nil
}

// 位于 lease/lessor.go:518
func (le *lessor) Attach(id LeaseID, items []LeaseItem) error {
 le.mu.Lock()
 defer le.mu.Unlock()
  // 从 LeaseMap 取出 LeaseID 对应的 lease
 l := le.leaseMap[id]
 if l == nil {
  return ErrLeaseNotFound
 }

 l.mu.Lock()
 for _, it := range items {
  l.itemSet[it] = struct{}{}
  le.itemMap[it] = id
 }
 l.mu.Unlock()
 return nil
}

// 位于 lease/lessor.go:311
func (le *lessor) Revoke(id LeaseID) error {
 le.mu.Lock()

 l := le.leaseMap[id]
 if l == nil {
  le.mu.Unlock()
  return ErrLeaseNotFound
 }
 defer close(l.revokec)
 // 释放锁
 le.mu.Unlock()

 if le.rd == nil {
  return nil
 }

 txn := le.rd()

 // 对键值进行排序，使得所有的成员保持删除键值对的顺序一致
 keys := l.Keys()
 sort.StringSlice(keys).Sort()
 for _, key := range keys {
  txn.DeleteRange([]byte(key), nil)
 }

 le.mu.Lock()
 defer le.mu.Unlock()
 delete(le.leaseMap, l.ID)
 // 键值删除操作需要在一个事务中进行
 le.b.BatchTx().UnsafeDelete(leaseBucketName, int64ToBytes(int64(l.ID)))

 txn.End()

 leaseRevoked.Inc()
 return nil
}

func testLease() {
    le := newLessor()    // 创建一个 Lessor
    le.Promote(0)        

    Go func() {   // 开启一个协程，接收过期的 key，主动删除
        for {  
           expireLease := <-le.ExpiredLeasesC()  

           for _, v := range expireLease {  
              le.Revoke(v.ID)    // 通过租约 ID 删除租约，删除租约时会从 backend 中删除绑定的 key
           }  
        }
    }()

    ttl = 5         
    lease := le.Grant(id, ttl)   // 申请一个租约

    le.Attach(lease, "foo")      // 将租约绑定在"foo"上

    time.Sleep(10 * time.Second)
}