您的位置:金沙游乐场85155 > 大数据库 > redis实现加锁的几种方法示例详解

redis实现加锁的几种方法示例详解

发布时间:2020-04-28 03:20编辑:大数据库浏览(101)

     $redis-setNX($key, $value); $redis-expire($key, $ttl);
    
    使用 SETNX 实现

    Redis 官方最早在 SETNX 命令页给了一个基于该命令的分布式锁实现

    1
    Acquire lock: SETNX lock.foo <current Unix time + lock timeout + 1>
    1
    Release lock: DEL lock.foo
    1. 如果 SETNX 返回 1,则表明客户端获取锁成功, lock.foo 被设置为有效 Unix time。客户端操作完成后调用 DEL 命令释放锁。

    2. 如果 SETNX 返回 0,则表明锁已经被其他客户端持有。这时我们可以先返回或进行重试等对方完成或等待锁超时。

    金沙游乐场85155,处理死锁问题:
    上述算法中,如果持有锁的客户端发生故障、意外崩溃、或者其他因素因素导致没有释放锁,该怎么解决?。我们可以通过锁的键对应的时间戳来判断这种情况是否发生了,如果当前的时间已经大于lock.foo的值,说明该锁已失效,可以被重新使用。
    发生这种情况时,可不能简单的通过DEL来删除锁,然后再SETNX一次,当多个客户端检测到锁超时后都会尝试去释放它,这里就可能出现一个竞态条件:

    1. C1 和 C2 读取 lock.foo 检查时间戳,先后发现超时了。
    2. C1 发送DEL lock.foo。
    3. C1 发送SETNX lock.foo 并且成功了。
    4. C2 发送DEL lock.foo
    5. C2 发送SETNX lock.foo 并且成功了。
    6. ERROR: 由于竞态的问题,C1 和 C2 都获取了锁,这下子问题大了。

    幸运的是,使用下面的算法可以避免这个问题。我们看看客户端 C4 是怎么做的:

    1. C4 发送 SETNX lock.foo 想要获取锁。
    2. 但是由于发生故障的客户端 C3 仍然持有锁,所以返回 0 给 C4。
    3. C4 发送 GET lock.foo 来检查锁是否过期, 如果没超时,则等待或重试。
    4. 反之,如果已经超时, C4 则尝试执行下面的命令来获取锁:

      1
      Acquire lock when time expired: GETSET lock.foo <current Unix timestamp + lock timeout + 1>
    5. 通过 GETSET ,C4 拿到的时间戳如果仍然是超时的,那就表明 C4 如愿以偿拿到锁了。

    6. 如果在 C4 之前,有个叫 C5 的客户端比 C4 快一步执行了上面的操作,那么 C4 拿到的时间戳是个未超时的值,这时,C4 没有如期获得锁,需要再次等待或重试。留意一下,尽管 C4 没拿到锁,但它改写了 C5 设置的锁的超时值,但是这点微小的误差(一般情况下锁的持有的时间非常短,所以在该竞态下出现的误差是可以容忍的)是可以容忍的。(Note that even if C4 set the key a bit a few seconds in the future this is not a problem)。

    为了这个锁的算法更健壮一些,持有锁的客户端在解锁之前应该再检查一次自己的锁是没有超时,再去做 DEL 操作,因为客户端失败的原因很复杂,不仅仅是崩溃也可能是因为某个耗时的操作而挂起,操作完的时候锁因为超时已经锁已经被别人获得,这时就不必解锁了。

    仔细的分析这个方案,我们就会发现这里有一个漏洞:Release lock 使用的 DEL 命令不支持 CAS 删除(check-and-set,delete if current value equals old value),在高并发情况下就会有一些问题:确认持有的锁没有超时后执行 DEL 释放锁,由于竞态的存在 Redis 服务器执行命令时锁可能已过期( “真的” 刚好过期或者被其他客户端竞争锁时设置了一个较小的过期时间而导致过期)且被其他客户端持有。这种情况下将会(非法)释放其他客户端持有的锁。

    解决方案: 先确定锁没有超时,再通过 EVAL 命令(在 Redis 2.6 及以上版本提供) 在执行 Lua 脚本:先执行 GET 指令获取锁的时间戳,确认和自己的时间戳一致后再执行 DEL 释放锁。

    设计缺陷:

    1. 上述解决方案并不完美,只解决了过期锁的释放问题,但是由于这个方案本身的缺陷,客户端获取锁时发生竞争(C4 改写 C5 时间戳的例子),那么 lock.foo 的 “时间戳” 将与本地的不一致,这个时候不会执行 DEL 命令,而是等待锁失效,这在高并发的环境下是低效的。
    2. 考虑多服务器环境下,需要服务器进行时间同步校准。

    在我们的项目中使用了 tornadoredis 库,这个库实现的分布式锁便采用了上述算法。但是在释放锁时有些限制,不过并发量不高的情况下不会有太大的问题,详细的分析参考下述代码注释。实现代码如下所示:

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89
    90
    91
    92
    93
    94
    95
    96
    97
    98
    99
    100
    101
    102
    103
    104
    105
    106
    107
    108
    109
    110
    111
    112
    113
    114
    115
    116
    117
    118
    119
    120
    121
    122
    123
    124
    125
    126
    127
    128
    129
    130
    131
    132
    133
    134
    135
    136
    137
    138
    139
    140
    141
    142
    143
    144
    class Lock(object):
        """
        A shared, distributed Lock that uses a Redis server to hold its state.
        This Lock can be shared across processes and/or machines. It works
        asynchronously and plays nice with the Tornado IOLoop.
        """
    
        LOCK_FOREVER = float(2 ** 31 + 1)  # 1 past max unix time
    
        def __init__(self, redis_client, lock_name, lock_ttl=None, polling_interval=0.1):
            """
            Create a new Lock object using the Redis key ``lock_name`` for
            state, that behaves like a threading.Lock.
            This method is synchronous, and returns immediately. It doesn't acquire the
            Lock or in fact trigger any sort of communications with the Redis server.
            This must be done using the Lock object itself.
            If specified, ``lock_ttl`` indicates the maximum life time for the lock.
            If none is specified, it will remain locked until release() is called.
            ``polling_interval`` indicates the time between acquire attempts (polling)
            when the lock is in blocking mode and another client is currently
            holding the lock.
            Note: If using ``lock_ttl``, you should make sure all the hosts
            that are running clients have their time synchronized with a network
            time service like ntp.
            """
            self.redis_client = redis_client
            self.lock_name = lock_name
            self.acquired_until = None
            self.lock_ttl = lock_ttl
            self.polling_interval = polling_interval
            if self.lock_ttl and self.polling_interval > self.lock_ttl:
                raise LockError("'polling_interval' must be less than 'lock_ttl'")
    
        @gen.engine
        def acquire(self, blocking=True, callback=None):
            """
            Acquire the lock.
            Returns True once the lock is acquired.
            If ``blocking`` is False, always return immediately. If the lock
            was acquired, return True, otherwise return False.
            Otherwise, block until the lock is acquired (or an error occurs).
            If ``callback`` is supplied, it is called with the result.
            """
    
            # Loop until we have a conclusive result
            while 1:
    
                # Get the current time
                unixtime = int(mod_time.time())
    
                # If the lock has a limited lifetime, create a timeout value
                if self.lock_ttl:
                    timeout_at = unixtime + self.lock_ttl
                # Otherwise, set the timeout value at forever (dangerous)
                else:
                    timeout_at = Lock.LOCK_FOREVER
                timeout_at = float(timeout_at)
    
                # Try and get the lock, setting the timeout value in the appropriate key,
                # but only if a previous value does not exist in Redis
                result = yield gen.Task(self.redis_client.setnx, self.lock_name, timeout_at)
    
                # If we managed to get the lock
                if result:
    
                    # We successfully acquired the lock!
                    self.acquired_until = timeout_at
                    if callback:
                        callback(True)
                    return
    
                # We didn't get the lock, another value is already there
                # Check to see if the current lock timeout value has already expired
                result = yield gen.Task(self.redis_client.get, self.lock_name)
                existing = float(result or 1)
    
                # Has it expired?
                if existing < unixtime:
    
                    # The previous lock is expired. We attempt to overwrite it, getting the current value
                    # in the server, just in case someone tried to get the lock at the same time
                    result = yield gen.Task(self.redis_client.getset,
                                            self.lock_name,
                                            timeout_at)
                    existing = float(result or 1)
    
                    # If the value we read is older than our own current timestamp, we managed to get the
                    # lock with no issues - the timeout has indeed expired
                    if existing < unixtime:
    
                        # We successfully acquired the lock!
                        self.acquired_until = timeout_at
                        if callback:
                            callback(True)
                        return
    
                    # However, if we got here, then the value read from the Redis server is newer than
                    # our own current timestamp - meaning someone already got the lock before us.
                    # We failed getting the lock.
    
                # If we are not signalled to block
                if not blocking:
    
                    # We failed acquiring the lock...
                    if callback:
                        callback(False)
                    return
    
                # Otherwise, we "sleep" for an amount of time equal to the polling interval, after which
                # we will try getting the lock again.
                yield gen.Task(self.redis_client._io_loop.add_timeout,
                               self.redis_client._io_loop.time() + self.polling_interval)
    
        @gen.engine
        def release(self, callback=None):
            """
            Releases the already acquired lock.
            If ``callback`` is supplied, it is called with True when finished.
            """
    
            if self.acquired_until is None:
                raise ValueError("Cannot release an unlocked lock")
    
            # Get the current lock value
            result = yield gen.Task(self.redis_client.get, self.lock_name)
            existing = float(result or 1)
    
            # 从上下文代码中可以看出,在这个实现中,有一个限制:获取锁的时候设置的 lock_ttl 必须能够保证释放锁时,锁未过期。
            # 否则,当前锁过期后,将会非法释放其他客户端持有的锁。如果无法估计持有锁后代码的执行时间,则可以增加当前锁的过期检测,
            # 当 self.acquired_until <= int(mod_time.time()) 时不执行 DEL 命令。不过,这个限制在一般的应用中倒是可以满足,
            # 所以这个实现不会有太大的问题。
            # 由于 GET、DEL 之间的时间差,以及 DEL 命令发出到 执行 之间的时间差,高并发情况下,锁过期释放的问题依然存在,这个是
            # 算法缺陷。并发不大的情况下,问题不大。
            #
            # 注:这个条件判断 existing >= self.acquired_until 是有这样一个潜在的前提,使用锁的客户端代码正常运行的情况下,
            # 考虑到并发代码使用相同的 lock_ttl 获取锁,竞争失败的客户端将会把锁的过期时间设置的更长一些,这里的判断是有意义的。
            # If the lock time is in the future, delete the lock
            if existing >= self.acquired_until:
                yield gen.Task(self.redis_client.delete, self.lock_name)
            self.acquired_until = None
    
            # That is it.
            if callback:
                callback(True)

    上面是官方提供的一个加锁方法,就是和第6的大体方法一样,只不过官方写的更健壮。所以可以直接使用官方提供写好的类方法进行调用。官方提供了各种语言如何实现锁。

    背景

    在一般的分布式应用中,要安全有效地同步多服务器多进程之间的共享资源访问,就要涉及到分布式锁。目前项目是基于 Tornado 实现的分布式部署,同时也使用了 Redis 作为缓存。参考了一些资料并结合项目自身的要求后,决定直接使用Redis实现全局的分布式锁。

    4、 客户端B在等待一段时间后在去请求设置key的值,设置成功

    单点问题

    上述实现都有一个单点问题: Redis 节点挂了肿么办?这是个很麻烦的问题,并且由于 Redis 主从复制是异步的,我们便不可能简单地实现互斥锁在节点间的安全迁移。当然一般的项目不会有这么高的要求,就目前我们的项目而言,本身Redis已经是单点。。。

    对于这个单点问题,Redis 上有一篇文章提供了一个算法来解决,但是实现比较复杂:

    《Distributed locks with Redis(原文)》 / 《使用 Redis 实现分布式锁(译文)》

    来源: http://strawhatfy.github.io/2015/07/09/Distributed%20locks%20with%20Redis/

    来自为知笔记(Wiz)

    上面两种方法都有一个问题,会发现,都需要设置 key 过期。那么为什么要设置key过期呢?如果请求执行因为某些原因意外退出了,导致创建了锁但是没有删除锁,那么这个锁将一直存在,以至于以后缓存再也得不到更新。于是乎我们需要给锁加一个过期时间以防不测。

    使用 SET resource-name anystring NX EX max-lock-time 实现

    该方案在 Redis 官方 SET 命令页有详细介绍。
    在介绍该分布式锁设计之前,我们先来看一下在从 Redis 2.6.12 开始 SET 提供的新特性,命令 SET key value [EX seconds] [PX milliseconds] [NX|XX],其中:

    1. EX seconds — 以秒为单位设置 key 的过期时间;
    2. PX milliseconds — 以毫秒为单位设置 key 的过期时间;
    3. NX — 将key 的值设为value ,当且仅当key 不存在,等效于 SETNX。
    4. XX — 将key 的值设为value ,当且仅当key 存在,等效于 SETEX。

    注:由于 SET 已经能够取代 SETNX, SETEX, PSETEX 命令,所以在未来的版本中,官方将逐渐放弃这3个命令,并最终移除。

    使用 SET 的新特性,改进旧版的分布式锁设计,主要有两个优化:

    1. 客户端通过 SET 命令可以同时完成获取锁和设置锁的过期时间:SET lock.foo token NX EX max-lock-time(原子操作,没有INCR 、 EXPIRE两个操作的事务问题),锁将在超时后自动过期,不担心之前设计的 “死锁” 问题,也没有多服务器时间同步校准的问题。

    2. 使用 Lua 脚本实现 CAS 删除,使锁更健壮。获取锁时为锁设置一个 token (一个无法猜测的随机字符串),释放锁时先比较 token 的值以保证只释放持有的有效锁。释放锁的 Lua 代码示例:

      1
      if redis.call("get",KEYS[1]) == ARGV[1]
      then
          return redis.call("del",KEYS[1])
      else
          return 0
      end

    这种加锁的思路是, key 不存在,那么 key 的值会先被初始化为 0 ,然后再执行 INCR 操作进行加一。然后其它用户在执行 INCR 操作进行加一时,如果返回的数大于 1 ,说明这个锁正在被使用当中。

    使用 Redis 实现分布式锁

    使用 Redis 实现分布式锁最简单方式是创建一对 key-value 值,key 被创建为有一定的生存期,因此它最终会被释放。而当客户端想要释放时,则直接删除 key 。基于不同的 Redis 命令,有两种实现方式:

    1. Redis 官方早期给的一个实现,使用 SETNX,将 value 设置为超时时间,由代码实现锁超时的检测[有缺陷,有限制,并发不高时可用];
    2. 有同学自己的实现:使用 INCR + EXPIRE,利用 Redis 的超时机制控制锁的生存期[不建议使用];
    3. Redis 官方给的一个改进实现:使用 SET resource-name anystring NX EX max-lock-time(Redis 2.6.12 后支持) 实现, 利用 Redis 的超时机制控制锁的生存期[Redis 2.6.12 以后建议使用]。

    针对问题2:针对第二个问题,在循环请求获取锁的时候,加入睡眠功能,等待几毫秒在执行循环

    使用 INCR + EXPIRE 实现

    该方案的实现来源这篇 blog 《Redis实现分布式全局锁》

    1. 客户端A通过 INCR locker.foo 获取名为 locker.foo 的锁,若获取的值为1,则表示获取成功,转入下一步,否则获取失败;
    2. 执行 EXPIRE locker.foo seconds 设置锁的过期时间,设置成功转入下一步;
    3. 执行共享资源访问;
    4. 执行 DEL locker.foo 释放锁。

    伪代码如下所示:

    1
    if(INCR('locker.foo') == 1)
    {
         // 设置锁的超时时间为1分钟,这个可以设置为一个较大的值来避免锁提前过期释放。
         EXPIRE(60)
    
         // 执行共享资源访问
         DO_SOMETHING()
    
         // 释放锁
         DEL('locker.foo')}
    }

    该实现有一个严重的“死锁”问题:如果 INCR 命令获取锁成功后,EXPIRE 失败,会导致锁无法正常释放。可用的解决方案是:借助 EVAL 命令,将 INCR 、 EXPIRE 操作封装在一个 Lua 脚本中执行,先执行 INCR 命令,成功获取锁后再执行 EXPIRE。以下是示例 Lua 代码:

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    -- Set a lock
    -- KEYS[1]   - key
    -- KEYS[2]   - ttl in ms
    
    local key     = KEYS[1]
    local ttl     = KEYS[2]
    
    local lockSet = redis.call('incr', key)
    
    if lockSet == 1 then
      redis.call('pexpire', key, ttl)
    end
    
    return lockSet

    注: 由于 EVAL 命令仅在 Redis 2.6 版本后提供,对于之前的版本只能通过 MULTI/EXEC 将 INCR 、 EXPIRE 封装在一个事务中来处理。但是由于 MULTI/EXEC 的限制,没有办法和使用 Lua 脚本一样根据 INCR 执行结果来执行 EXPIRE ,所以如果获取锁失败,会导致 TTL 不断被延长,在高并发的环境里如果拿到锁的进程意外挂掉而没有正常释放锁,锁便只能等到过期才能被其他客户端持有,而这个过期时间的长短取决于获取锁时的竞争激烈情况。该解决方案有严重缺陷,不适合高并发环境

    ++实际上,由于不能通过一个原语完成获取锁和设置锁过期时间的操作,即使通过上述 Lua 脚本来获取锁,仍然是有问题的。由于 Redis 事务的特点,只保证 INCR 、 EXPIRE 两条命令在 Redis 上是连续执行的,但当 EXPIRE 命令失败后并不会回滚 INCR 命令,所以 “死锁” 问题依然没有解决(取决于 Redis 的稳定性)。同时,也存在锁过期后非法释放其他客户端持有的锁的问题,且由于依赖 redis 的自动过期机制,便无法检测到此问题。++

    2、 客户端B也去请求服务器获取key的值为2表示获取锁失败

     do { //针对问题1,使用循环 $timeout = 10; $roomid = 10001; $key = 'room_lock'; $value = 'room_'.$roomid; //分配一个随机的值针对问题3 $isLock = Redis::set($key, $value, 'ex', $timeout, 'nx');//ex 秒 if ($isLock) { if (Redis::get($key) == $value) { //防止提前过期,误删其它请求创建的锁 //执行内部代码 Redis::del($key); continue;//执行成功删除key并跳出循环 } } else { usleep(5000); //睡眠,降低抢锁频率,缓解redis压力,针对问题2 } } while(!$isLock);
    

    针对问题3:在加锁的时候存入的key是随机的。这样的话,每次在删除key的时候判断下存入的key里的value和自己存的是否一样

    6. 解决办法

    1、 客户端A请求服务器获取key的值为1表示获取了锁

    5、 客户端B执行代码完成,删除锁

    3、 客户端A执行代码完成,删除锁

     $redis-incr($key); $redis-expire($key, $ttl); //设置生成时间为1秒
    

    1、 客户端A请求服务器设置key的值,如果设置成功就表示加锁成功

     $servers = [ ['127.0.0.1', 6379, 0.01], ['127.0.0.1', 6389, 0.01], ['127.0.0.1', 6399, 0.01], ]; $redLock = new RedLock($servers); //加锁 $lock = $redLock-lock('my_resource_name', 1000); //删除锁 $redLock-unlock($lock)
    

    1、 客户端A请求服务器设置key的值,如果设置成功就表示加锁成功

    虽然上面一步已经满足了我们的需求,但是还是要考虑其它问题?

    3. 第二种锁SETNX

    2、 循环请求的话,如果有一个获取了锁,其它的在去获取锁的时候,是不是容易发生抢锁的可能?

    3、 锁提前过期后,客户端A还没执行完,然后客户端B获取到了锁,这时候客户端A执行完了,会不会在删锁的时候把B的锁给删掉?

    3、 客户端A执行代码完成,删除锁

    4、 客户端B在等待一段时间后在去请求设置key的值,设置成功

    本文由金沙游乐场85155发布于大数据库,转载请注明出处:redis实现加锁的几种方法示例详解

    关键词:

上一篇:redis安装和配置_动力节点Java学院整理

下一篇:没有了