10.Redis锁代码

# 01.Redis分布式锁

# 1.1 Redis事物介绍

# 1、事物介绍

1.Redis事物是可以一次执行多个命令，本质是一组命令的集合
2.一个事务中的所有命令都会序列化，按顺序串行化的执行而不会被其他命令插入
**作用：**一个队列中，一次性、顺序性、排他性的执行一系列命令

# Multi 命令用于标记一个事务块的开始事务块内的多条命令会按照先后顺序被放进一个队列当中，最后由 EXEC 命令原子性( atomic )地执行
> multi        # 开始一个Redis事物
incr books
incr books
> exec         # 执行事物
> discard       # 丢弃事物

1
2
3
4
5
6

Redis事务提供以下几个命令

MULTI # 标记一个事务块的开始
EXEC # 执行所有事务块内的命令
DISCARD # 取消事务，放弃执行事务块内的所有命令
WATCH # 监视一个或多个key，如果在事务执行前这些key的值发生了改变，事务将被打断

1
2
3
4

# 2、如何解决 Redis 事务的缺陷

Redis 从 2.6 版本开始支持执行 Lua 脚本，它的功能和事务非常类似
我们可以利用 Lua 脚本来批量执行多条 Redis 命令
一段 Lua 脚本可以视作一条命令执行
一段 Lua 脚本执行过程中不会有其他脚本或 Redis 命令同时执行，保证了操作不会被其他指令插入或打扰
不过，如果 Lua 脚本运行时出错并中途结束，出错之后的命令是不会被执行的
并且，出错之前执行的命令是无法被撤销的，无法实现类似关系型数据库执行失败可以回滚的那种原子性效果
因此，严格来说的话，通过 Lua 脚本来批量执行 Redis 命令实际也是不完全满足原子性的
如果想要让 Lua 脚本中的命令全部执行，必须保证语句语法和命令都是对的

# 1.2 watch乐观锁

实质：WATCH 只会在数据被其他客户端抢先修改了的情况下通知执行命令的这个客户端（通过 WatchError 异常）但不会阻止其他客户端对数据的修改
1）原理
- 1.当用户购买时，通过 WATCH 监听用户库存，如果库存在watch监听后发生改变，就会捕获异常而放弃对库存减一操作
- 2.如果库存没有监听到变化并且数量大于1，则库存数量减一，并执行任务
2）弊端
- 1.Redis 在尝试完成一个事务的时候，可能会因为事务的失败而重复尝试重新执行
- 2.保证商品的库存量正确是一件很重要的事情，但是单纯的使用 WATCH 这样的机制对服务器压力过大

使用reids的 watch + multi 指令实现超卖

#! /usr/bin/env python
# -*- coding: utf-8 -*-
import Redis
def sale(rs):
    while True:
        with rs.pipeline() as p:
            try:
                p.watch('apple')                      # 监听key值为apple的数据数量改变
                count = int(rs.get('apple'))
                print('拿取到了苹果的数量: %d' % count)
                p.multi()                           # 事务开始
                if count> 0 :                        # 如果此时还有库存
                    p.set('apple', count - 1)
                    p.execute()                      # 执行事务
                p.unwatch()
                break                              # 当库存成功减一或没有库存时跳出执行循环
            except Exception as e:                       # 当出现watch监听值出现修改时，WatchError异常抛出
                print('[Error]: %s' % e)
                continue                            # 继续尝试执行

rs = Redis.Redis(host='127.0.0.1', port=6379)                   # 连接Redis
rs.set('apple',1000)                                    # 首先在Redis中设置某商品apple 对应数量value值为1000
sale(rs)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# 1.3 setnx分布式锁

1.分布式锁本质是占一个坑，当别的进程也要来占坑时发现已经被占，就会放弃或者稍后重试
2.占坑一般使用 setnx(set if not exists)指令，只允许一个客户端占坑
3.先来先占，用完了在调用del指令释放坑

> setnx lock:codehole true
.... do something critical ....
> del lock:codehole

1
2
3

# 1、分布式锁解决超买

setnx+watch+multi解决超卖问题

#! /usr/bin/env python
# -*- coding: utf-8 -*-
import Redis
import uuid
import time

# 1.初始化连接函数
def get_conn(host,port=6379):
    rs = Redis.Redis(host=host, port=port)
    return rs

# 2. 构建Redis锁
def acquire_lock(rs, lock_name, expire_time=10):
    '''
    rs: 连接对象
    lock_name: 锁标识
    acquire_time: 过期超时时间
    return -> False 获锁失败 or True 获锁成功
    '''
    identifier = str(uuid.uuid4())
    end = time.time() + expire_time
    while time.time() < end:
        # 当获取锁的行为超过有效时间，则退出循环，本次取锁失败，返回False
        if rs.setnx(lock_name, identifier): # 尝试取得锁
            return identifier
        time.sleep(0.001)
        return False

# 3. 释放锁
def release_lock(rs, lockname, identifier):
    '''
    rs: 连接对象
    lockname: 锁标识
    identifier: 锁的value值，用来校验
    '''
    pipe = rs.pipeline(True)
    try:
        pipe.watch(lockname)
        if rs.get(lockname).decode() == identifier:  # 防止其他进程同名锁被误删
            pipe.multi()           # 开启事务
            pipe.delete(lockname)
            pipe.execute()
            return True            # 删除锁
        pipe.unwatch()              # 取消事务
    except Exception as e:
        pass
    return False                    # 删除失败


'''在业务函数中使用上面的锁'''
def sale(rs):
    start = time.time()            # 程序启动时间
    with rs.pipeline() as p:
        '''
        通过管道方式进行连接
        多条命令执行结束，一次性获取结果
        '''
        while True:
            lock = acquire_lock(rs, 'lock')
            if not lock: # 持锁失败
                continue
            try:
                count = int(rs.get('apple')) # 取量
                p.set('apple', count-1)      # 减量
                p.execute()
                print('当前库存量: %s' % count)
                break
            finally:
                release_lock(rs, 'lock', lock)
        print('[time]: %.2f' % (time.time() - start))

rs = Redis.Redis(host='127.0.0.1', port=6379)      # 连接Redis
rs.set('apple',1000)                               # 首先在Redis中设置某商品apple 对应数量value值为1000
sale(rs)

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

# 2、优化

给分布式锁加超时时间防止死锁

def acquire_expire_lock(rs, lock_name, expire_time=10, locked_time=10):
    '''
    rs: 连接对象
    lock_name: 锁标识
    acquire_time: 过期超时时间
    locked_time: 锁的有效时间
    return -> False 获锁失败 or True 获锁成功
    '''
    identifier = str(uuid.uuid4())
    end = time.time() + expire_time
    while time.time() < end:
        # 当获取锁的行为超过有效时间，则退出循环，本次取锁失败，返回False
        if rs.setnx(lock_name, identifier): # 尝试取得锁
            # print('锁已设置: %s' % identifier)
            rs.expire(lock_name, locked_time)
            return identifier
        time.sleep(.001)
    return False

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

# 02.Redis分布式锁不可靠

# 1、Redis分布式锁不可靠产生的原因

锁超时： 线程 A 获取了锁，并设置了超时时间，由于执行时间长，导致锁的超时时间到期
锁释放：线程 A 的锁超时，系统自动将锁释放，但是此时线程 A 的业务逻辑可能还没有执行完成
其他线程获取锁： 在锁被释放的瞬间，其他线程 B 获取了锁
线程 A继续执行： 线程 A 完成后释放锁，由于锁已经被线程 B 获取，线程 A 误释放了线程 B 持有的锁

# 2、解决误释放其他线程锁的方法

加锁时指定锁标识
是否锁时判断当前持有的锁是否是自己的锁标识

import Redis
import uuid
import time

class DistributedLock:
    def __init__(self, Redis_host='localhost', Redis_port=6379, Redis_db=0, lock_key='distributed_lock'):
        self.Redis_conn = Redis.StrictRedis(host=Redis_host, port=Redis_port, db=Redis_db)
        self.lock_key = lock_key
        self.lock_value = None

    def acquire_lock(self, timeout=10):
        identifier = str(uuid.uuid4())  # 锁标识
        end_time = time.time() + timeout  # 过期超时时间

        while time.time() < end_time:
            # 当获取锁的行为超过有效时间，则退出循环，本次取锁失败，返回False
            if self.Redis_conn.set(self.lock_key, identifier, nx=True, ex=timeout):
                self.lock_value = identifier
                return True
            time.sleep(0.1)

        return False

    def release_lock(self):
        # Redis.call("get", KEYS[1]) 获取锁当前的标识符
        # ARGV[1]是当前标识符，如果相同则表示这是持有锁的正确客户端
        # 如果标识符不同，返回 0 表示释放锁失败，因为这个客户端不是持有锁的正确客户端
        # 使用 Lua 脚本可以确保上述操作是原子性的，从而避免了竞态条件
        lua_script = """
            if Redis.call("get", KEYS[1]) == ARGV[1] then
                return Redis.call("del", KEYS[1])
            else
                return 0
            end
        """
        try:
            self.Redis_conn.eval(lua_script, 1, self.lock_key, self.lock_value)
        except Redis.exceptions.RedisError:
            pass


if __name__ == "__main__":
    # 使用示例
    lock = DistributedLock()

    try:
        if lock.acquire_lock():
            # 执行需要加锁的操作
            print("Lock acquired successfully")
            time.sleep(5)
        else:
            print("Failed to acquire lock within timeout")
    except Exception as e:
        print(f"An error occurred: {e}")
    finally:
        lock.release_lock()
        print("Lock released")

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

# 3、抢购业务中如何解决此问题

进入业务后业务先将库存减1存入缓存（就是先假设商品能卖成功）
分布式锁不仅判断锁也判断库存，如果库存不够也不准进入
.举例：
- 比如库存10个，每个请求进来后都将缓存库存-1，这时12个请求过来，第11、12个是进不来的
- 但这个时候卖失败两个，缓存库存+2
- 后面的第13、14个请求就能进来了要注意的是库存+-要是安全的
主要利用 Redis 在哈希表中的字段上执行增加或减少操作的原子性来实现

# 创建一个名为 key:info 的哈希表，并设置其中的 stock 字段的初始值为 10
HSET key:info count 10
# 对名为 key:info 的哈希表中的 count 字段递减 1操作
HINCRBY key:info count -1

1
2
3
4

# 03.看门狗（Watchdog）

Redis 分布式锁可以引入看门狗机制（Watchdog），通过动态延长锁的过期时间，确保在业务逻辑执行期间，锁不会过期

看门狗机制的核心思想是通过定时续约的方式来防止锁的过早失效

加锁操作： Redis 加锁方式是使用 SET key value NX PX expire_time，其中 expire_time 是初始的过期时间
启动看门狗：
- 当客户端成功获取到锁后，会启动一个后台线程（看门狗），这个线程定期检查锁是否仍然持有
- 如果仍然持有，就会向 Redis 发送续约请求（延长锁的过期时间）
- 这样，锁的过期时间在看门狗的管理下动态延长，直到业务逻辑执行完毕
定期续约：
- 看门狗线程会以一个小于锁过期时间的间隔（比如每隔一半的过期时间）发送命令来延长锁的 TTL（存活时间）
- 例如，如果锁的初始过期时间是 30 秒，客户端每隔 15 秒就会向 Redis 发送续约请求，将锁的过期时间重新设置为 30 秒
释放锁： 当业务逻辑执行完成后，客户端显式地调用 DEL 命令释放锁，同时停止看门狗线程

# 04.Redlock

Redlock 的主要作用是在多个 Redis 实例上实现分布式锁

确保在网络分区、部分节点不可用、网络延迟等复杂分布式环境下，仍能提供一种安全且高效的分布式锁机制

其核心思想是在多节点 Redis 集群中通过一定的策略，确保锁只被一个客户端持有，从而避免多个客户端同时执行临界区代码

# 1、Redlock 工作原理

Redlock 的工作流程大致分为以下几个步骤

部署多实例 Redis：
- 首先需要在不同的物理机器或虚拟机上部署多个 Redis 实例（至少 3 个，一般建议使用 5 个 Redis 实例）
- 这些实例互相独立，不需要主从同步，也不需要复制数据，它们只作为独立的锁存储单元
获取锁：
- 客户端生成一个UUID，然后依次向多个 Redis 实例发送 SET resource_name UUID NX PX ttl 命令请求加锁
- NX 确保锁定资源时，只有资源未被其他客户端占用才会成功
- PX ttl 设置锁的过期时间，防止死锁
- 锁请求的超时时间通常较短（如 10 毫秒），以避免因网络延迟导致的阻塞
判断锁是否成功：
- 客户端必须至少在 半数以上的 Redis 实例 上成功获取锁（假设是 5 个实例，则至少需要在 3 个实例上获取到锁）
- 只有当客户端能够在一个合理的时间窗口内（比如锁的有效期）成功获取到多数锁时，才认为加锁成功
计算是否加锁成功：
- 客户端在多个 Redis 实例上获取锁，并且获取过程总时长小于锁的 TTL 时，认为成功获取了全局锁
- 若加锁失败，客户端应及时释放已经获取的锁，避免资源被长时间占用
释放锁：
- 当客户端完成业务逻辑后，会通过 DEL 命令主动释放在每个 Redis 实例上的锁

# 2、Redlock 正确性保证

防止网络分区问题：
- 即使某些 Redis 实例因为网络问题或故障无法响应请求
- 客户端依然可以通过与多数 Redis 实例的交互来确保锁的安全性
容错性：
- Redlock 机制要求客户端必须在大多数实例上获取锁
- 这意味着即使部分节点发生故障（如有一个节点宕机），也不会影响锁的安全性
锁的自动失效：
- Redis 提供的 TTL 机制确保即使客户端崩溃，锁也会在超时时间到期后自动释放，防止死锁
锁的唯一性：
- 每次获取锁时，客户端生成的 UUID 确保即使在多个 Redis 实例上获取同一资源的锁，每个客户端持有的锁都是唯一的

# 3、Redlock 使用场景

分布式任务调度：
- 在分布式环境中，多个工作节点可能会并发执行同一个任务
- 通过 Redlock，确保只有一个节点能执行任务，其他节点会被阻塞
电商系统中的订单处理：
- 在电商系统中，多个用户可能会同时尝试购买库存有限的商品
- 通过 Redlock，可以确保每个商品的库存更新是原子性的，避免出现超卖的情况
分布式缓存更新：
- 在分布式缓存中，多个服务实例可能会同时更新同一个缓存条目
- 通过 Redlock，确保缓存更新操作的顺序性和唯一性，避免出现数据不一致问题
跨区域分布式系统：
- 在跨区域的分布式系统中，由于网络延迟和分区问题，可能会导致多个节点同时持有锁
- 使用 Redlock，可以在网络延迟较大的环境中确保锁的安全性和一致性

# 4、Redlock 的局限性

网络延迟问题：
- Redlock 依赖于客户端在多个实例上的时间窗口判断锁的有效性
- 若网络延迟较大，可能导致客户端误判锁的获取状态
Redis 实例故障：
- 如果在加锁过程中，Redis 实例大规模宕机，Redlock 可能无法及时释放锁，影响其他客户端获取锁
时间一致性假设：
- Redlock 机制假设 Redis 实例之间的时钟同步是大致一致的，这在大规模分布式环境中可能不完全成立

← 09.雪崩穿透击穿 11.Redis缓存→