Redis入门--头歌实验分布式锁与信号量

港控/mmm° 2024-04-17 13:49 38阅读 0赞

分布式锁，是指在分布式的部署环境下，通过锁机制来让多客户端**互斥的**对共享资源进行访问。

信号量的本质也是一种数据操作锁，它本身不具有数据交换的功能，而是通过控制其他的通信资源来实现进程间通信，从而负责数据操作的互斥与同步。

## 一、简易锁 ##

> ##### 任务描述 #####
> 
> 由 `WATCH`、`MULTI` 和 `EXEC` 命令组成的事务并**不具备可扩展性**，简易锁提供了一种可扩展的并发控制机制。
> 
> 本关任务：使用 `Redis`构建简易锁。
> 
> ##### 相关知识 #####
> 
> 为了完成本关任务，你需要掌握：`1`．redis基本命令，`2`．`python`基本命令。
> 
> ###### redis基本命令 ######
> 
> ###### setnx： ######
> 
> **当且仅当** `key` 不存在时，为 `key` 赋值。可通过 setnx 实现互斥锁。
> 
>     # 创建一个连接到Redis服务器的对象
>     conn = redis.Redis()
>     
>     # 尝试设置键为"key:exists"的值为"changed"，如果键已存在则不设置
>     conn.setnx("key:exists", "changed")
>     
>     # 尝试设置键为"key:new"的值为"hello"，如果键已存在则不设置
>     conn.setnx("key:new", "hello")
> 
> 执行前：
> 
> `key:exists` `constant`
> 
> 执行后：
> 
> `key:new` `hello` `key:exists` `constant`
> 
> 执行结果：
> 
>     False
>     True
> 
> ###### del： ######
> 
> 删除指定的 `key`，不存在的 `key` 会被忽略。
> 
>     conn = redis.Redis()
>     conn.delete("key:exists")
>     conn.delete("key:not_exists")
> 
> 执行前：
> 
> `key:other` `value_b` `key:exists` `value_a`
> 
> 执行后：
> 
> `key:other` `value_b`
> 
> ###### watch： ######
> 
> 用于监视一个 key ，如果在事务执行之前这个 key 的值被改动，那么事务将被打断。其可以配合 unwatch 命令一同使用。
> 
> 例子：商品下单时是需要对库存数`stock_count`进行减库存操作，通过`watch`和`unwatch`来锁库存。
> 
>     conn = redis.Redis()
>     pipe = conn.pipeline()
>     try:
>     pipe.watch('stock_count')
>     count = int(pipe.get('stock_count'))
>     if count > 0:  # 有库存
>     pipe.set('stock_count', count - 1)
>     return count
>     pipe.unwatch() # 取消所有的监视
>     except redis.exceptions.WatchError: # 抓取stock_count改变异常
> 
> ###### python基本命令 ######
> 
> 返回当前时间的时间戳
> 
>     time.time()
> 
> 使进程休眠`0.5`秒：
> 
>     time.sleep(0.5)
> 
> 返回全局唯一标识符：
> 
>     str(uuid.uuid4())
> 
> 执行结果：`eda1c452-5f53-422b-a2f7-577f2ad43c14`
> 
> 判断字符串是否相等（一致）：
> 
>     'same_string' == 'same_string'
>     'same_string' == 'other_string'
> 
> 执行结果：
> 
>     True
>     False
> 
> ##### 编程要求 #####
> 
> 在`Begin-End`区域编写 `acquire_lock(lockname, acquire_timeout=5)` 函数，实现获得锁的功能，具体参数与要求如下：
> 
>  *  方法参数 `lockname` 是要获取的锁的名字，`acquire_timeout` 是获取该锁的最长等待时间；
>  *  获得锁的实现：在等待时间内，不断尝试使用 `setnx` 命令将该锁对应的键`lock:{lockname}`的值设置为一个全局唯一标识符
>     
>      *  若成功，为方便之后释放对应的锁，返回该唯一标识；
>      *  若失败，为给其他客户端释放该锁的时间，先休眠`0.001`秒，若超时仍未获得该锁，则返回`False`。
> 
> 编写 `release_lock(lockname, identifier)` 函数，实现获得锁的功能，具体参数与要求如下：
> 
>  *  方法参数 `lockname` 是要释放的锁的名字，`identifier` 是指定的唯一标识；
>  *  释放锁的实现：若要释放的锁的唯一标识与指定的唯一标识一致，则删除该锁对应的键`lock:{lockname}`，并返回`True`；若不一致，则返回`False`；
>  *  为了避免重复释放同一个锁，使用`watch`监控`lock:{lockname}`，如果`lock:{lockname}`已经修改，则捕获`redis.exceptions.WatchError`，然后重试释放该锁。

#!/usr/bin/env python
    #-*- coding:utf-8 -*-
    
    import uuid
    import time
    import redis
    
    conn = redis.Redis()
    
    # 获得锁
    def acquire_lock(lockname, acquire_timeout=5):
        # 生成一个唯一的标识符
        identifier = str(uuid.uuid4())
        # 计算锁的超时时间
        end = time.time() + acquire_timeout
        # 循环尝试获取锁，直到超时
        while time.time() < end:
            # 使用setnx命令尝试设置锁，如果成功则返回标识符
            if conn.setnx('lock:' + lockname, identifier):
                return identifier
            # 如果失败则等待0.001秒后再次尝试
            time.sleep(0.001)
        # 如果超时仍未获得锁，则返回False
        return False
    
    # 释放锁
    def release_lock(lockname, identifier):
        # 创建一个管道对象
        pipe = conn.pipeline()
        # 拼接锁的名称
        lockname = 'lock:' + lockname
        # 循环尝试释放锁，直到成功或失败
        while True:
            try:
                # 监视锁的名称
                pipe.watch(lockname)
                # 如果锁的值等于标识符，则删除锁并返回True
                if pipe.get(lockname) == identifier:
                    pipe.delete(lockname)
                    return True
                # 如果锁的值不等于标识符，则取消监视并退出循环
                pipe.unwatch()
                break
            # 如果发生WatchError异常，则继续尝试
            except redis.exceptions.WatchError:
                pass
        # 如果无法释放锁，则返回False
        return False

## 二、超时限制锁 ##

> ##### 任务描述 #####
> 
> 本关任务：使用 `Redis`构建超时限制锁。
> 
> ##### 相关知识 #####
> 
> 上一实验中构建的锁只是基本上正确。一旦客户端在获得锁之后发生崩溃，就会导致锁**一直处于**“被占用”的状态，进而导致死锁。这一实验中，我们将为锁加上超时功能。
> 
> 为了完成本关任务，你需要掌握：`1`、`redis`相关命令，`2`、`python`相关命令。
> 
> ###### redis相关命令 ######
> 
> ###### setnx： ######
> 
> **当且仅当** `key` 不存在时，为 `key` 赋值。可通过 setnx 实现互斥锁。
> 
>     conn = redis.Redis()
>     conn.setnx("key:exists", "changed")
>     conn.setnx("key:new", "hello")
> 
> 执行前：
> 
> `key:exists` `constant`
> 
> 执行后：
> 
> `key:new` `hello` `key:exists` `constant`
> 
> 执行结果：
> 
>     执行前：
>     
>     key:exists constant
>     
>     执行后：
>     
>     key:new hello
>     key:exists constant
>     
>     执行结果：
> 
> ###### expire： ######
> 
> 为给定 `key` 设置生存时间，当 `key` 过期时（生存时间为 `0`），会被自动删除。
> 
>     conn = redis.Redis()
>     conn.expire("testkey", 2)
> 
> 等待两秒后，`testkey` 键会被自动删除。
> 
> ###### ttl： ######
> 
> 返回 `key` 的剩余生存时间，单位为秒。
> 
>     conn = redis.Redis()
>     conn.set("non_durable_key", "1")
>     conn.expire("non_durable_key", 10)
>     conn.ttl("non_durable_key")
> 
> 执行结果：`10.0`
> 
> 当键未设置过生存时间时：
> 
>     conn = redis.Redis()
>     conn.set("durable_key", "2)
>     conn.ttl("durable_key")
> 
> 执行结果：`None`
> 
> ###### python相关命令 ######
> 
> 返回当前时间的时间戳
> 
>     time.time()
> 
> 使进程休眠`0.5`秒：
> 
>     time.sleep(0.5)
> 
> 返回全局唯一标识符：
> 
>     str(uuid.uuid4())
> 
> 执行结果：`eda1c452-5f53-422b-a2f7-577f2ad43c14`
> 
> 判断两个数字的大小：
> 
>     3 < 0
>     None < 0
> 
> 执行结果：
> 
>     False
>     True
> 
> ##### 编程要求 #####
> 
> 在`Begin-End`区域编写 `acquire_lock_with_timeout(lockname, acquire_timeout=10, lock_timeout=10)` 函数，实现获得超时限制锁的功能，具体参数与要求如下：
> 
>  *  方法参数`lockname`是要获取的锁的名字，`acquire_timeout`是获得该锁的最大等待时间，`lock_timeout`是该锁的过期时间；
>  *  获得锁的实现：在等待时间内，不断尝试使用 `setnx` 命令将该锁对应的键`lock:{lockname}`的值设置为一个全局唯一标识符
>     
>      *  若成功，则设置锁的过期时间，并**返回**该唯一标识；
>      *  若失败，为了避免锁的过期时间未设置上，需要检测锁的剩余生存时间，若不存在，则重新设置锁的过期时间。若超时仍未获得该锁，则返回`False`。

#!/usr/bin/env python
    #-*- coding:utf-8 -*-
    
    import uuid
    import time
    import redis
    
    conn = redis.Redis()
    
    # 获得超时限制锁
    def acquire_lock_with_timeout(lockname, acquire_timeout=10, lock_timeout=10):
        # 生成一个唯一的标识符
        identifier = str(uuid.uuid4())
        # 将锁名加上前缀'lock:'
        lockname = 'lock:' + lockname
        # 计算获取锁的截止时间
        end = time.time() + acquire_timeout
        # 在规定时间内尝试获取锁
        while time.time() < end:
            # 如果成功设置键值对，则返回唯一标识符
            if conn.setnx(lockname, identifier):
                conn.expire(lockname, lock_timeout)
                return identifier
            # 如果锁已经过期，则重新设置过期时间
            elif conn.ttl(lockname) < 0:
                conn.expire(lockname, lock_timeout)
            # 等待一段时间后再次尝试获取锁
            time.sleep(0.001)
        # 超过规定时间仍未获取到锁，返回False
        return False

## 三、计数信号量 ##

> ##### 任务描述 #####
> 
> 本关任务：使用 `Redis`构建计数信号量。
> 
> ##### 相关知识 #####
> 
> 计数信号量也是**一种锁**，限定了能够同时使用的资源数量，从而限制了某个关键代码段的最大并发量。
> 
> 为了完成本关任务，你需要掌握：`1`、`redis`相关命令，`2`、`python`相关命令。
> 
> ###### redis相关命令 ######
> 
> ###### zremrangebyscore： ######
> 
> 移除有序集合中指定分值区间内的所有成员。
> 
>     conn = redis.Redis()
>     conn.zremrangebyrank("testzset", '-inf', 100)
> 
> 执行前：
> 
> 1.  `member1 20.0`
> 2.  `member2 135.0`
> 3.  `member3 200.0`
> 
> 执行后：
> 
> 1.  `member2 135.0`
> 2.  `member3 200.0`
> 
> ###### zadd： ######
> 
> 将成员加入到有序集合中，并确保其在正确的位置上。
> 
>     conn = redis.Redis()
>     conn.zadd("testzset", "member2", 3)
>     conn.zadd("testzset", "member1", 2)
>     conn.zadd("testzset", "member3", 1)
> 
> 执行后：
> 
> 1.  `member3`
> 2.  `member1`
> 3.  `member2`
> 
> ###### zrank： ######
> 
> 返回有序集合中指定成员的排名。
> 
>     conn = redis.Redis()
>     conn.zrank("testzset", "member2")
> 
> 执行结果：`2`
> 
> zrem：从有序集合中移除指定成员。
> 
>     conn = redis.Redis()
>     conn.zrem("testzset", "member1")
> 
> 执行前：
> 
> 1.  `member3`
> 2.  `member1`
> 3.  `member2`
> 
> 执行后：
> 
> 1.  `member3`
> 2.  `member2`
> 
> 执行结果：`1`
> 
> ###### pipeline： ######
> 
> 使用管道批量提交命令，其一般配合 execute 使用。一般用于减少通信往返开销。
> 
> 例子：为了减少通信往返开销，使用 `pipeline` 和 `execute` 批量提交命令：
> 
>     conn = redis.Redis()
>     pipe = conn.pipeline()
>     pipe.incrby("testcount", 1)
>     pipe.lpush("testlist", "member1")
>     print(pipe.execute()[0])
> 
> 两条命令会一次性提交到服务器，并一同返回结果，可以通过元组下标获取到单个结果。
> 
> 执行结果：`1`
> 
> ###### python相关命令 ######
> 
> 返回当前时间的时间戳。
> 
>     time.time()
> 
> 返回全局唯一标识符：
> 
>     str(uuid.uuid4())
> 
> 执行结果：`eda1c452-5f53-422b-a2f7-577f2ad43c14`
> 
> ##### 编程要求 #####
> 
> 在`Begin-End`区域编写 `acquire_semaphore(semname, limit=5, timeout=10)` 函数，实现获得计数信号量的功能，具体参数与要求如下：
> 
>  *  方法参数`semname`是信号量存储的有序集合键名，`limit`是最大的信号量个数，`timeout`是信号量的有效时间；
>  *  该类有序集合中存储了已获得的信号量：成员为信号量的唯一标识，分值为信号量的获得时间，其结构如下：
> 
> ![91d7cea0c7b74009b52a4a979afc2110.png][]
> 
> ![7e48fd2960974fed941660b5f368e290.png][]

#!/usr/bin/env python
    #-*- coding:utf-8 -*-
    
    import uuid
    import time
    import redis
    
    conn = redis.Redis()
    
    # 获得计数信号量
    def acquire_semaphore(semname, limit=5, timeout=10):
        # 生成一个唯一的标识符
        identifier = str(uuid.uuid4())
        # 获取当前时间戳
        now = time.time()
        # 创建一个管道对象
        pipeline = conn.pipeline()
        # 移除过期的信号量
        pipeline.zremrangebyscore(semname, '-inf', now - timeout)
        # 添加新的信号量
        pipeline.zadd(semname, identifier, now)
        # 获取新添加的信号量的排名
        pipeline.zrank(semname, identifier)
        # 如果排名小于限制数，则返回标识符，否则移除该信号量并返回None
        if pipeline.execute()[-1] < limit:
            return identifier
    
        conn.zrem(semname, identifier)
        return None
    
    # 释放计数信号量
    def release_semaphore(semname, identifier):
        # 从信号量集合中移除指定的信号量
        return conn.zrem(semname, identifier)

## 四、公平信号量 ##

> ##### 任务描述 #####
> 
> 本关任务：使用 `Redis`实现公平信号量。
> 
> ##### 相关知识 #####
> 
> 当各个客户端（应用服务器）的系统时间不一致时，计数信号量会出现分配紊乱问题。为了减少不正确的系统时间对获得信号量操作的影响，需要使用 `Redis`统一产生的值进行信号量的排名。
> 
> 为了完成本关任务，你需要掌握：`1`、`redis`相关命令，`2`、`python`相关命令。
> 
> ###### redis相关命令 ######
> 
> ###### zremrangebyscore： ######
> 
> 移除有序集合中指定分值区间内的所有成员。
> 
>     conn = redis.Redis()
>     conn.zremrangebyrank("testzset", '-inf', 100)
> 
> 执行前：
> 
> 1.  `member1 20.0`
> 2.  `member2 135.0`
> 3.  `member3 200.0`
> 
> 执行后：
> 
> 1.  `member2 135.0`
> 2.  `member3 200.0`
> 
> ###### zinterstore： ######
> 
> 根据权重计算两个有序集合的**交集**，并将结果存放在指定的键中。
> 
>     conn = redis.Redis()
>     conn.zinterstore("sum_point", {"mid_test": 0.3, "fin_test": 0.7})
> 
> 执行前：
> 
> ![ec39542d8e2b4915a3ffd3d941391e1d.png][]
> 
> 执行后：
> 
> ![0d4ab2c0f3ca4b54b529d28877224646.png][]
> 
> ###### incr： ######
> 
> 将 `key` 中储存的数字值增一。
> 
>     conn = redis.Redis()
>     conn.incr("testcount")
> 
> 执行前：不存在 `testcount` 键。
> 
> 执行后：
> 
> `testcount` `'1'`
> 
> ###### zadd： ######
> 
> 将成员加入到有序集合中，并确保其在正确的位置上。
> 
>     conn = redis.Redis()
>     conn.zadd("testzset", "member2", 3)
>     conn.zadd("testzset", "member1", 2)
>     conn.zadd("testzset", "member3", 1)
> 
> 执行后：
> 
> 1.  `member3`
> 2.  `member1`
> 3.  `member2`
> 
> ###### zrank： ######
> 
> 返回有序集合中指定成员的排名。
> 
>     conn = redis.Redis()
>     conn.zrank("testzset", "member2")
> 
> 执行结果：`2`
> 
> ###### zrem： ######
> 
> 从有序集合中移除指定成员。
> 
>     conn = redis.Redis()
>     conn.zrem("testzset", "member1")
> 
> 执行前：
> 
> 1.  `member3`
> 2.  `member1`
> 3.  `member2`
> 
> 执行后：
> 
> 1.  `member3`
> 2.  `member2`
> 
> 执行结果：`1`
> 
> ###### python相关命令 ######
> 
> 返回当前时间的时间戳。
> 
>     time.time()
> 
> 返回全局唯一标识符：
> 
>     str(uuid.uuid4())
> 
> 执行结果：`eda1c452-5f53-422b-a2f7-577f2ad43c14`

#!/usr/bin/env python
    #-*- coding:utf-8 -*-
    
    import uuid
    import time
    import redis
    
    conn = redis.Redis()
    
    # 获得公平信号量
    def acquire_fair_semaphore(semname, limit=5, timeout=10):
        # 生成一个唯一的标识符
        identifier = str(uuid.uuid4())
        czset = semname + ':owner'
        ctr = semname + ':counter'
    
        now = time.time()
        pipeline = conn.pipeline(True)
        # 移除超时的标识符
        pipeline.zremrangebyscore(semname, '-inf', now - timeout)
        # 将当前时间之前的标识符从原集合中移除，并添加到新集合中
        pipeline.zinterstore(czset, {czset: 1, semname: 0})
    
        # 计数器加一
        pipeline.incr(ctr)
        counter = pipeline.execute()[-1]
    
        # 将标识符和当前时间添加到原集合中
        pipeline.zadd(semname, identifier, now)
        # 将标识符和计数器的值添加到新集合中
        pipeline.zadd(czset, identifier, counter)
    
        # 获取标识符在新集合中的排名
        pipeline.zrank(czset, identifier)
        # 如果排名小于限制数，则返回标识符，否则移除标识符并返回None
        if pipeline.execute()[-1] < limit:
            return identifier
    
        pipeline.zrem(semname, identifier)
        pipeline.zrem(czset, identifier)
        pipeline.execute()
        return None
    
    # 释放公平信号量
    def release_fair_semaphore(semname, identifier):
        # 从原集合和新集合中移除指定的标识符
        pipeline = conn.pipeline()
        pipeline.zrem(semname, identifier)
        pipeline.zrem(semname + ':owner', identifier)
        return pipeline.execute()[0]

[91d7cea0c7b74009b52a4a979afc2110.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/13/085e22eab891466ab7996c32838dbda2.png
[7e48fd2960974fed941660b5f368e290.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/13/3dac9f12163142388e1173628a9ec556.png
[ec39542d8e2b4915a3ffd3d941391e1d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/13/1d24dffb77d64afa8efec46f54c844b5.png
[0d4ab2c0f3ca4b54b529d28877224646.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/13/591a2026efc84385a31971983d917b4d.png