教你如何使用Redis实现分布式锁的方法

for i := 0; i < m.tries; i++ {
if i != 0 {
select {
case <-ctx.Done():
// Exit early if the context is done.
return ErrFailed
case <-time.After(m.delayFunc(i)):
// Fall-through when the delay timer completes.
}
}
start := time.Now()
// 尝试在所有的节点中加锁
n, err := func() (int, error) {
ctx, cancel := context.WithTimeout(ctx, time.Duration(int64(float64(m.expiry)*m.timeoutFactor)))
defer cancel()
return m.actOnPoolsAsync(func(pool redis.Pool) (bool, error) {
// acquire 加锁函数
return m.acquire(ctx, pool, value)
})
}()
if n == 0 && err != nil {
return err
}
// 如果加锁节点书没有达到的设定的数目
// 或者键值的过期时间已经到了
// 在所有的节点中解锁
now := time.Now()
until := now.Add(m.expiry – now.Sub(start) – time.Duration(int64(float64(m.expiry)*m.driftFactor)))
if n >= m.quorum && now.Before(until) {
m.value = value
m.until = until
return nil
}
_, err = func() (int, error) {
ctx, cancel := context.WithTimeout(ctx, time.Duration(int64(float64(m.expiry)*m.timeoutFactor)))
defer cancel()
return m.actOnPoolsAsync(func(pool redis.Pool) (bool, error) {
// 解锁函数
return m.release(ctx, pool, value)
})
}()
if i == m.tries-1 && err != nil {
return err
}
}
return ErrFailed
}
// 遍历所有的节点，并且在每个节点中执行传入的函数
func (m *Mutex) actOnPoolsAsync(actFn func(redis.Pool) (bool, error)) (int, error) {
type result struct {
Status bool
Err error
}
ch := make(chan result)
// 执行传入的函数
for _, pool := range m.pools {
go func(pool redis.Pool) {
r := result{}
r.Status, r.Err = actFn(pool)
ch <- r
}(pool)
}
n := 0
var err error
// 计算执行成功的节点数目
for range m.pools {
r := <-ch
if r.Status {
n++
} else if r.Err != nil {
err = multierror.Append(err, r.Err)
}
}
return n, err
}
// 手动解锁的lua脚本
var deleteScript = redis.NewScript(1, `
if redis.call(“GET”, KEYS[1]) == ARGV[1] then
return redis.call(“DEL”, KEYS[1])
else
return 0
end
`)
// 手动解锁
func (m *Mutex) release(ctx context.Context, pool redis.Pool, value string) (bool, error) {
conn, err := pool.Get(ctx)
if err != nil {
return false, err
}
defer conn.Close()
status, err := conn.Eval(deleteScript, m.name, value)
if err != nil {
return false, err
}
return status != int64(0), nil
}

// ExtendContext resets the mutex’s expiry and returns the status of expiry extension.
func (m *Mutex) ExtendContext(ctx context.Context) (bool, error) {
start := time.Now()
// 尝试在所有的节点中加锁
n, err := m.actOnPoolsAsync(func(pool redis.Pool) (bool, error) {
return m.touch(ctx, pool, m.value, int(m.expiry/time.Millisecond))
})
if n < m.quorum {
return false, err
}
// 判断下锁的过期时间
now := time.Now()
until := now.Add(m.expiry – now.Sub(start) – time.Duration(int64(float64(m.expiry)*m.driftFactor)))
if now.Before(until) {
m.until = until
return true, nil
}
return false, ErrExtendFailed
}

var touchScript = redis.NewScript(1, `
// 需要先比较下当前的value值
if redis.call(“GET”, KEYS[1]) == ARGV[1] then
return redis.call(“PEXPIRE”, KEYS[1], ARGV[2])
else
return 0
end
`)

func (m *Mutex) touch(ctx context.Context, pool redis.Pool, value string, expiry int) (bool, error) {
conn, err := pool.Get(ctx)
if err != nil {
return false, err
}
defer conn.Close()
status, err := conn.Eval(touchScript, m.name, value, expiry)
if err != nil {
return false, err
}
return status != int64(0), nil
}

found = (lookupKeyWrite(c->db,key) != NULL);
// 这里是 SETEX 实现的重点
// 如果nx,并且在数据库中找到了这个值就返回
// 如果是 xx,并且在数据库中没有找到键值就会返回

// 因为 Redis 中的命令执行都是单线程操作的
// 所以命令中判断如果存在就返回，能够保证正确性，不会出现并发访问的问题
if ((flags & OBJ_SET_NX && found) ||
(flags & OBJ_SET_XX && !found))
{
if (!(flags & OBJ_SET_GET)) {
addReply(c, abort_reply ? abort_reply : shared.null[c->resp]);
}
return;
}

…
}