分布式锁总结

func (l *Lock) Lock(stopCh <-chan struct{}) (<-chan struct{}, error) {
  // Hold the lock as we try to acquire
  l.l.Lock()
  defer l.l.Unlock()

  // Check if we already hold the lock
  if l.isHeld {
    return nil, ErrLockHeld
  }

  // Check if we need to create a session first
  // 如果没有session，创建一个，用于标识获取锁的对象和过期管理
  l.lockSession = l.opts.Session
  if l.lockSession == "" {
    s, err := l.createSession()
    if err != nil {
      return nil, fmt.Errorf("failed to create session: %v", err)
    }

    l.sessionRenew = make(chan struct{})
    l.lockSession = s
    session := l.c.Session()
    // 在没释放锁之前一直刷新session的时间
    go session.RenewPeriodic(l.opts.SessionTTL, s, nil, l.sessionRenew)

    // If we fail to acquire the lock, cleanup the session
    defer func() {
      if !l.isHeld {
    close(l.sessionRenew)
    l.sessionRenew = nil
      }
    }()
  }

  // Setup the query options
  kv := l.c.KV()
  qOpts := &QueryOptions{
    WaitTime: l.opts.LockWaitTime,
  }

  start := time.Now()
  attempts := 0
WAIT:
  // Check if we should quit
  select {
  case <-stopCh:
    return nil, nil
  default:
  }

  // Handle the one-shot mode.
  if l.opts.LockTryOnce && attempts > 0 {
    elapsed := time.Since(start)
    if elapsed > qOpts.WaitTime {
      return nil, nil
    }

    qOpts.WaitTime -= elapsed
  }
  attempts++

  // Look for an existing lock, blocking until not taken
  pair, meta, err := kv.Get(l.opts.Key, qOpts)
  if err != nil {
    return nil, fmt.Errorf("failed to read lock: %v", err)
  }
  if pair != nil && pair.Flags != LockFlagValue {
    return nil, ErrLockConflict
  }
  locked := false
  if pair != nil && pair.Session == l.lockSession {
    goto HELD
  }
  if pair != nil && pair.Session != "" {
    qOpts.WaitIndex = meta.LastIndex
    goto WAIT
  }

  //获取一个lock
  pair = l.lockEntry(l.lockSession)
  locked, _, err = kv.Acquire(pair, nil)
  if err != nil {
    return nil, fmt.Errorf("failed to acquire lock: %v", err)
  }

  // Handle the case of not getting the lock
  if !locked {
    // Determine why the lock failed
    qOpts.WaitIndex = 0
    pair, meta, err = kv.Get(l.opts.Key, qOpts)
    if pair != nil && pair.Session != "" {
      //If the session is not null, this means that a wait can safely happen
      //using a long poll
      qOpts.WaitIndex = meta.LastIndex
      goto WAIT
    } else {
      // If the session is empty and the lock failed to acquire, then it means
      // a lock-delay is in effect and a timed wait must be used
      select {
      case <-time.After(DefaultLockRetryTime):
    goto WAIT
      case <-stopCh:
    return nil, nil
      }
    }
  }

HELD:
  // Watch to ensure we maintain leadership
  leaderCh := make(chan struct{})
  //一直监视锁的占用者，当锁的占用者不是当前session，close掉leaderCh
  go l.monitorLock(l.lockSession, leaderCh)

  // Set that we own the lock
  l.isHeld = true

  // Locked! All done
  return leaderCh, nil
}

//unlock操作比较简单
func (l *Lock) Unlock() error {
  // Hold the lock as we try to release
  l.l.Lock()
  defer l.l.Unlock()

  // Ensure the lock is actually held
  if !l.isHeld {
    return ErrLockNotHeld
  }

  // Set that we no longer own the lock
  l.isHeld = false

  // Stop the session renew
  // 停止延长session的时间
  if l.sessionRenew != nil {
    defer func() {
      close(l.sessionRenew)
      l.sessionRenew = nil
    }()
  }

  // Get the lock entry, and clear the lock session
  lockEnt := l.lockEntry(l.lockSession)
  l.lockSession = ""

  // Release the lock explicitly
  //直接调用release释放掉lock
  kv := l.c.KV()
  _, _, err := kv.Release(lockEnt, nil)
  if err != nil {
    return fmt.Errorf("failed to release lock: %v", err)
  }
  return nil
}