从原理到实践，分析 Redis 分布式锁的多种实现方案（一）

末蓝、 2024-03-16 18:27 122阅读 0赞

一、为什么要用分布式锁

在分布式系统中，为了保证多个进程或线程之间的数据一致性和正确性，需要使用锁来实现互斥访问共享资源。然而，使用本地锁在分布式系统中存在问题。

本地锁的问题

  1. 无法保证全局唯一性：本地锁只在本地生效，每个节点都有自己的一份数据，所以不能保证在整个集群中全局唯一。
  2. 无法协调多个节点之间的锁：在分布式系统中，多个节点同时访问同一个资源时，需要协调各个节点之间的锁，保证资源的互斥访问。而本地锁只能锁住当前节点的资源，无法协调各个节点之间的锁。
  3. 可能会出现死锁和锁竞争：由于分布式系统中很难保证各个节点的锁同步，因此容易导致死锁和锁竞争等问题。
  4. 性能问题：在分布式系统中，为了保证多个节点之间的锁同步，通常需要进行大量的网络通信，这会影响系统的性能。

比如商品服务A和服务B同时获取到库存数量为10的商品信息。商品服务A和服务B同时进行扣减库存操作，分别将库存数量减少了1。商品服务A和服务B均修改了库存数量为9，然后将数据写入数据库中。
由于使用本地锁，商品服务A和服务B之间没有进行协调，因此就会出现数据不一致的问题。可能出现以下情况：商品服务A先将库存数量9写入数据库，然后商品服务B也将库存数量9写入数据库，商品服务B先将库存数量9写入数据库，然后商品服务A也将库存数量9写入数据库，结果，整个系统中库存数量实际只完成了一次扣减，最终库存数量卖出2份后，还剩下9，出现了数据不一致的情况。

8ace7d8a6d1b47d484098ccd27f382af.png

相比之下，分布式锁可以解决上述问题。分布式锁可以在多个节点之间协调锁的使用，确保在分布式系统中多个进程或线程互斥访问共享资源，并保证了全局唯一性，避免了死锁和锁竞争问题，同时也能够提高系统的吞吐量和性能。

二、什么是分布式锁

分布式锁是一种用于在分布式系统中协调多个进程或线程之间对共享资源的互斥访问的机制。在分布式系统中，由于各个节点之间没有共享内存，因此无法使用传统的本地锁机制来实现进程或线程的同步，所以需要使用分布式锁来解决这个问题。

举一个生活中的例子，假设我们去乘坐高铁，首先要进行检票进站，但有很多人都想进站。为了避免大家同时挤进去，高铁站会设置检票闸机，每次只允许一人检票通过，当有人检票进入时，其他人必须等待，直到检票成功进入后，闸机会再次反锁。后面的人再尝试检票获取检票闸机的进入权。这里的检票闸机就是高铁站的一把锁。

065fee9e5e9045008521b918ddfeb5f7.png

来看下分布式锁的基本原理，如下图所示：

0249d1a93ccf4b0e9a05feb40a169905.png

我们来分析下上图的分布式锁：

  • 1.前端将 100个 的高并发请求转发两个商品微服务。
  • 2.每个微服务处理 50个请求。
  • 3.每个处理请求的线程在执行业务之前，需要先抢占锁。可以理解为“占坑”。
  • 4.获取到锁的线程在执行完业务后，释放锁。可以理解为“释放坑位”。
  • 5.未获取到的线程需要等待锁释放。
  • 6.释放锁后，其他线程抢占锁。
  • 7.重复执行步骤 4、5、6。

大白话解释：所有请求的线程都去同一个地方“占坑”，如果有坑位，就执行业务逻辑，没有坑位，就需要其他线程释放“坑位”。这个坑位是所有线程可见的，可以把这个坑位放到 Redis 缓存或者数据库，这篇讲的就是如何用 Redis 做“分布式坑位”。

分布式锁的好处

  1. 避免重复操作：如果多个进程或线程同时尝试对同一个资源进行操作，就会导致重复操作和数据的不一致。使用分布式锁可以确保只有一个进程或线程能够获得锁，从而避免了重复操作。
  2. 防止竞态条件：在并发环境下，多个进程或线程同时读写共享资源时，容易引发竞态条件（Race Condition）。使用分布式锁可以保证同一时间只有一个进程或线程能够访问共享资源，从而避免了竞态条件。
  3. 提高系统吞吐量：使用分布式锁可以避免多个进程或线程同时竞争共享资源，从而有效地提高系统的吞吐量和性能。

三、Redis 的 SETNX

为了使用分布式锁，需要我们找到一个可靠的第三方中间件。Redis刚好可以用来作为分布式锁的提供者。

主要原因在于 Redis 具有以下特点：

  1. 高性能：Redis 是一种内存数据库，数据存储在内存中，读写速度非常快，可以快速响应锁的获取和释放请求。
  2. 原子操作：Redis 支持原子操作，例如 SETNX（SET if Not eXists）命令可以实现“只有在键不存在时设置键值”的操作，可以保证同时只会有一个客户端成功获取到锁，并且避免了因为执行多个操作而导致的竞态条件问题。
  3. 可靠性高：Redis 可以进行主从复制和持久化备份等操作，可以确保即使出现网络中断或 Redis 实例宕机的情况，也可以保证分布式锁的正确性和一致性。

基于以上特点，我们可以使用 Redis 来实现分布式锁的机制。具体做法是通过 SETNX 命令在 Redis 中创建一个键值对作为锁，当有其他客户端尝试获取锁时，如果该键值对已经存在，则表示锁已经被其他客户端持有；反之，则表示当前客户端获取锁成功。

Redis 中的 SETNX 命令用于设置指定键的值，但是只有在该键不存在时才进行设置。如果该键已经存在，则 SETNX 命令不会对其进行任何操作。

SETNX 的语法如下：

  1. SETNX key value

SETNX 的源码实现比较简单，其实现过程如下：

  1. 检查给定键是否在 Redis 中已经存在，如果存在则返回 0，不对 key 的值进行修改。
  2. 如果 key 不存在，则将 key 的值设置为 value，并返回 1。

SETNX 命令的 C 语言实现如下：

  1. void setnxCommand(client *c) {
  2.     robj *o;
  3.     int nx = c->argc == 3; /* 如果参数个数为 3，说明设置 NX（key 不存在才设置） */
  4.     long long expire = 0; /* 默认不设置过期时间 */
  5.     int retval;
  7.     if (nx) {
  8.         /* NX 模式下检查 key 是否已经存在 */
  9.         if (lookupKeyWrite(c->db,c->argv[1]) != NULL) {
  10.             addReply(c,shared.czero);
  11.             return;
  12.         }
  13.     } else {
  14.         /* XX 模式下检查 key 是否不存在 */
  15.         if (lookupKeyWrite(c->db,c->argv[1]) == NULL) {
  16.             addReply(c,shared.czero);
  17.             return;
  18.         }
  19.     }
  21.     /* 尝试将字符串型或整型数字转换为 long long 型数字 */
  22.     if (getTimeoutFromObjectOrReply(c,c->argv[3],&expire,UNIT_SECONDS)
  23.         != C_OK) return;
  24.     /* 值为空则返回错误 */
  25.     if (checkStringLength(c,c->argv[2]->ptr,sdslen(c->argv[2]->ptr)) != C_OK)
  26.         return;
  27.     /* 尝试将键值对插入到数据库中 */
  28.     o = createStringObject(c->argv[2]->ptr,sdslen(c->argv[2]->ptr));
  29.     retval = dictAdd(c->db->dict,c->argv[1],o);
  30.     if (retval == DICT_OK) {
  31.         incrRefCount(o);
  32.         /* 设置过期时间 */
  33.         if (expire) setExpire(c->db,c->argv[1],mstime()+expire);
  34.         server.dirty++;
  35.         addReply(c, shared.cone);
  36.     } else {
  37.         decrRefCount(o);
  38.         addReply(c, shared.czero);
  39.     }
  40. }

从源码实现可以看出，SETNX 命令的执行过程非常快速，由于 Redis 存储数据是采用字典结构，在判断 key 是否存在时可以达到 O(1) 的时间复杂度，因此 SETNX 命令的性能很高。

四、使用Redis SETNX 实现分布式锁的方案

SETNX 方案流程图

c849df07cb5c4f8fa3c7c25d174aaba6.png

如上图所示，使用 Redis 的 SETNX 命令来实现分布式锁的过程如下：

  1. 客户端尝试获取锁，以锁的名称为键名，将客户端唯一标识（如 UUID）作为键值，调用 Redis 的 SETNX 命令。
  2. 如果 Redis 中不存在该键，即返回的结果是 1，则表示锁获取成功，客户端可以进入临界区进行操作。
  3. 如果 Redis 中已经存在该键，即返回的结果是 0，则表示锁已经被其他客户端持有，当前客户端没有获取到锁，需要等待或重试。
  4. 当客户端完成操作后，调用 Redis 的 DEL 命令来释放锁，删除键。

代码示例

  1. @Service
  2. public class ProductService {
  4.     private final RedisTemplate<String, String> redisTemplate;
  6.     @Autowired
  7.     public ProductService(RedisTemplate<String, String> redisTemplate) {
  8.         this.redisTemplate = redisTemplate;
  9.     }
  11.     /**
  12.      * 扣减库存
  13.      *
  14.      * @param productId 商品ID
  15.      * @param quantity  数量
  16.      * @return true 扣减成功，false 扣减失败
  17.      */
  18.     public boolean decreaseStock(String productId, int quantity) {
  19.         String lockKey = "stock_" + productId;
  20.         while (true) {
  21.             Boolean lockResult = redisTemplate.opsForValue().setIfAbsent(lockKey, "", 10, TimeUnit.SECONDS);
  22.             if (lockResult != null && lockResult) {
  23.                 try {
  24.                     String stockKey = "product_" + productId;
  25.                     String stockStr = redisTemplate.opsForValue().get(stockKey);
  26.                     if (StringUtils.isEmpty(stockStr)) {
  27.                         // 库存不存在或已过期
  28.                         return false;
  29.                     }
  30.                     int stock = Integer.parseInt(stockStr);
  31.                     if (stock < quantity) {
  32.                         // 库存不足
  33.                         return false;
  34.                     }
  35.                     int newStock = stock - quantity;
  36.                     redisTemplate.opsForValue().set(stockKey, String.valueOf(newStock));
  37.                     return true;
  38.                 } finally {
  39.                     redisTemplate.delete(lockKey);
  40.                 }
  41.             }
  42.             try {
  43.                 Thread.sleep(10000);
  44.             } catch (InterruptedException e) {
  45.                 Thread.currentThread().interrupt();
  46.             }
  47.         }
  48.     }
  49. }

在 decreaseStock() 方法中，首先定义了一个 lockKey，用于对商品库存进行加锁。进入 while 循环后，使用 Redis 的 setIfAbsent() 方法尝试获取锁，如果返回值为 true，则表示成功获取锁。在成功获取锁后，再从 Redis 中获取商品库存，判断库存是否充足，如果充足则扣减库存并返回 true；否则直接返回 false。最后，在 finally 块中删除加锁的 key。

如果获取锁失败，则等待 10 秒后再次尝试获取锁，直到获取成功为止。

SETNX 实现分布式锁的缺陷

使用 Redis SETNX 实现分布式锁可能存在以下缺陷：

  1. 竞争激烈时容易出现死锁情况。这种情况可以通过在加锁时设置一个唯一标识符（例如 UUID），释放锁时检查标识符是否匹配来避免。
  2. 锁的释放不及时。可以通过在加锁时设置一个过期时间，确保即使客户端意外宕机，锁也会在一定时间后自动释放。
  3. 客户端误删其他客户端的锁。这种情况可以通过为每个客户端生成一个唯一标识符，加锁时将标识符写入 Redis，释放锁时检查标识符是否匹配来避免。

五、Redis SETNX优化方案 SETNXEX

针对使用 Redis SETNX 实现分布式锁可能出现死锁的情况,，可以使用SETNXEX进行优化，Redis SETNXEX 命令是 Redis 提供的一个原子操作指令，用于设置一个有过期时间的字符串类型键值对，当且仅当该键不存在时设置成功，返回 1，否则返回 0。SETNXEX 命令的语法如下：

  1. SETNXEX key seconds value

其中，key 是键名；seconds 为整数，表示键值对的过期时间（单位为秒）；value 是键值。

源码分析：

实现 SETNXEX 命令的关键在于如何保证该操作的原子性和一致性。其实现过程如下：

  1. 如果键 key 已经存在，则返回 0。
  2. 如果键 key 不存在，则将键 key 的值设置为 value，并设置过期时间为 seconds 秒。如果设置成功，则返回 1；否则，返回 0。

Redis 在底层使用 SETNX 和 SETEX 命令实现 SETNXEX 命令，它的 C 语言实现代码如下：

  1. void setnxexCommand(client *c) {
  2.     robj *key = c->argv[1], *val = c->argv[3];
  3.     long long expire = strtoll(c->argv[2]->ptr,NULL,10);
  4.     expire *= 1000;
  6.     if (getExpire(c,key) != -1) {
  7.         addReply(c, shared.czero);
  8.         return;
  9.     }
  10.     setKey(c,c->db,key,val,LOOKUP_NOTOUCH|LOOKUP_EX|LOOKUP_NX,0,0,NULL);
  11.     if (c->flags & CLIENT_MULTI) {
  12.         addReply(c, shared.cone);
  13.         return;
  14.     }
  15.     server.dirty++;
  16.     if (expire) setExpire(c,c->db,key,mstime()+expire);
  17.     addReply(c, shared.cone);
  18.     notifyKeyspaceEvent(NOTIFY_STRING,"set",key,c->db->id);
  19.     notifyKeyspaceEvent(NOTIFY_GENERIC,"expire",key,c->db->id);
  20. }

在这个代码中，首先从客户端传来的参数中获取 key、value 和 expire 值，并通过 getExpire 函数检查键是否已经存在。如果已经存在，则返回 0；否则，调用 setKey 函数将键值对设置为 value，并加上过期时间 expire。当然，这里的过期时间是以毫秒为单位的，需要转换成 Redis 的标准格式。最后，通过 addReply 函数向客户端发送成功的响应消息，并通过 notifyKeyspaceEvent 函数发送键空间通知。

需要注意的是，虽然 SETNXEX 被称为“原子操作”，但实际上在高并发场景下，SETNX 和 SETEX 操作之间可能会发生竞争问题，导致 SETNX 和 SETEX 操作不具备原子性。如果在分布式场景下需要保证 SETNXEX 的原子性，还需要使用分布式锁等机制来避免竞争问题。因此，在使用 SETNXEX 命令时，需要根据具体情况，评估其安全性和可靠性，采用合适的解决方案。

六、使用Redis SETNXEX 实现分布式锁的方案

SETNXEX 方案流程图

eee541e65a88479980f15b6f810f51c7.png

如上图所示，使用 Redis 的 SETNXEX 命令来实现分布式锁的过程如下

  1. 客户端向 Redis 服务器发送申请锁的请求，请求内容包括锁的名称和过期时间；
  2. Redis 服务器接收到请求后进行处理，使用 SETNXEX 命令将锁键和值写入到 Redis 的键值对数据库中，并设置过期时间；
  3. 如果 SETNXEX 返回值是 1，则客户端成功获取到锁，执行业务逻辑并在完成后释放锁；
  4. 如果 SETNXEX 返回值是 0，则客户端未获取到锁，等待一段时间后重试获取锁；
  5. 客户端在释放锁时，先确认自己是否持有该锁，如果持有则使用 DEL 命令删除锁。

代码示例

  1. @Component
  2. public class StockService {
  3.     private final Logger logger = LoggerFactory.getLogger(StockService.class);
  4.     private final String LOCK_KEY_PREFIX = "stock:lock:";
  5.     @Autowired
  6.     private RedisTemplate<String, Object> redisTemplate;
  8.     /**
  9.      * 扣减库存
  10.      * @param productId 商品ID
  11.      * @param num 扣减数量
  12.      */
  13.     public boolean reduceStock(Long productId, int num) {
  14.         // 构造锁的key
  15.         String lockKey = LOCK_KEY_PREFIX + productId;
  16.         // 构造锁的value，这里使用当前线程的ID
  17.         String lockValue = String.valueOf(Thread.currentThread().getId());
  18.         try {
  19.             // 尝试获取锁，设置过期时间为10秒
  20.             Boolean locked = redisTemplate.opsForValue().setIfAbsent(lockKey, lockValue, 10L, TimeUnit.SECONDS);
  21.             if (!locked) {
  22.                 // 获取锁失败，等待10秒后重新尝试获取锁
  23.                 Thread.sleep(10000);
  24.                 return reduceStock(productId, num);
  25.             }
  26.             // 获取锁成功，执行扣减库存代码
  27.             // TODO ... 扣减库存代码
  28.             return true;
  29.         } catch (InterruptedException e) {
  30.             logger.error("Failed to acquire stock lock", e);
  31.             return false;
  32.         } finally {
  33.             // 释放锁
  34.             if (lockValue.equals(redisTemplate.opsForValue().get(lockKey))) {
  35.                 redisTemplate.delete(lockKey);
  36.             }
  37.         }
  38.     }
  39. }

上述代码中，首先构造了锁的key和value，然后使用 RedisTemplate 的 setIfAbsent 方法尝试获取锁。如果获取锁失败，则线程会等待10秒后重新尝试获取锁，直到获取锁成功为止。如果获取锁成功，则执行扣减库存的业务逻辑，待操作完成后释放锁。

SETNXEX 实现分布式锁的缺陷

  1. 非阻塞式获取锁：使用 SETNXEX 命令获取锁时，如果锁已经被其他客户端持有，则 SETNXEX 操作会失败，并返回 0。在这种情况下，当前客户端可以继续执行其他操作，而无需等待锁的释放。这种非阻塞式获取锁的策略可能会导致死锁和数据竞争问题，对系统的可靠性和正确性产生负面影响。
  2. 锁过期机制：使用 SETNXEX 命令设置锁时需要指定过期时间，如果锁的持有者在过期时间内没有完成操作，锁会自动释放，从而导致其他客户端可以获取该锁。但是，如果在锁过期前持有锁的客户端还未完成操作，那么其他客户端就有可能获取到该锁，从而导致多个客户端同时修改同一个资源，引发数据竞争问题。
  3. 非可重入锁：使用 SETNXEX 命令获取锁时，不能重复获取已经持有的锁，否则会导致死锁问题。因此，SETNXEX 命令实现的分布式锁是一种非可重入锁，不能满足某些场景下的需求。
  4. 非原子性操作：在分布式环境中，如果在比较锁的值和删除锁之间，有其他客户端获取了锁并修改了数据，那么该锁的值可能已经被改变，导致误删锁或删除其他客户端持有的锁，引发数据竞争问题。

七、Redis SETNXEX 实现分布式锁缺陷的优化方案

针对SETNXEX锁过期问题的优化方案：在执行业务逻辑前，我们设置锁的过期时间为 30 秒，并启动一个定时任务续租锁，以防止锁因长时间持有而超时失效。
在 finally 块中释放锁，首先判断当前线程是否持有该锁，如果是则删除该锁。

SETNXEX 优化方案流程图

" class="reference-link">00b1624aa5ea4f418e37a5b7b2126c2a.png

如上图所示，当有两个线程同时请求获取锁时，执行流程如下：

  1. 线程 A 和线程 B 同时想要获取名为 lock 的锁。
  2. 线程 A 先到达 Redis 中，执行 SETNXEX lock 30 命令尝试获取锁。如果返回值为 1，则说明线程 A 成功获取到锁，进入业务逻辑执行阶段。
  3. 线程 B 到达 Redis 中，执行 SETNXEX lock 30 命令尝试获取锁。由于线程 A 已经获取了锁且正在执行业务逻辑，因此线程 B 获取锁失败，需要等待一段时间后重新尝试获取。
  4. 在获取锁失败后，线程 B 进入等待状态，等待一段时间后再次尝试获取锁。
  5. 在线程 A 执行业务逻辑前，将锁的过期时间设置为 30 秒，并开启一个定时任务每隔 10 秒续租一次锁，以保证在业务逻辑执行期间锁不会超时失效。
  6. 在 finally 块中释放锁，首先判断当前线程是否持有该锁，如果是则删除该锁。如果线程 A 的业务逻辑执行完毕，则释放锁；如果线程 B 成功获取到锁，并在后面的某个时间释放了锁，之后的请求会有机会获取到锁。

代码示例

  1. @Component
  2. public class StockService {
  3.     @Autowired
  4.     private RedisTemplate<String, String> redisTemplate;
  6.     /**
  7.      * 扣减库存
  8.      *
  9.      * @param stockId 库存 ID
  10.      * @param num 扣减数量
  11.      * @return 是否扣减成功
  12.      */
  13.     public boolean reduceStock(String stockId, int num) throws InterruptedException {
  14.         // 构造锁的名称
  15.         String lockKey = "stock_lock_" + stockId;
  16.         // 获取当前线程 ID
  17.         String threadId = String.valueOf(Thread.currentThread().getId());
  19.         try {
  20.             // 使用 SETNXEX 命令申请锁
  21.             Boolean lockResult = redisTemplate.opsForValue().setIfAbsent(lockKey, threadId, 30, TimeUnit.SECONDS);
  23.             if (!lockResult) {
  24.                 // 如果获取锁失败，则等待一段时间后重试
  25.                 Thread.sleep(10000);
  26.                 return reduceStock(stockId, num);
  27.             }
  29.             // 设置锁的过期时间为 30 秒，并启动一个定时任务续租锁
  30.             ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
  31.             scheduledExecutorService.scheduleAtFixedRate(() -> {
  32.                 Long expireResult = redisTemplate.getExpire(lockKey);
  33.                 if (expireResult < 10) {
  34.                     redisTemplate.expire(lockKey, expireResult + 10, TimeUnit.SECONDS);
  35.                 }
  36.             }, 10, 10, TimeUnit.SECONDS);
  38.             // TODO：执行业务逻辑，例如扣减库存
  40.             return true;
  41.         } catch (Exception e) {
  42.             e.printStackTrace();
  43.         } finally {
  44.             // 释放锁
  45.             if (threadId.equals(redisTemplate.opsForValue().get(lockKey))) {
  46.                 redisTemplate.delete(lockKey);
  47.             }
  48.         }
  50.         return false;
  51.     }
  52. }

在上面的代码示例中，我们使用了 redisTemplate 的 opsForValue().setIfAbsent() 方法来申请锁。如果获取锁失败，则等待 10 秒后重新尝试获取锁。在获取锁成功后，我们设置锁的过期时间为 30 秒，并启动一个定时任务续租锁，以防止锁因长时间持有而超时失效。在执行完业务逻辑后，返回 true 表示扣减成功。

在释放锁时，我们首先通过 redisTemplate.opsForValue().get(lockKey) 方法获取当前持有锁的线程 ID，然后判断当前线程是否持有该锁，如果是则删除该锁。这里使用了 redisTemplate.delete() 方法来删除锁。

方案的缺陷

  1. 可重入性问题：如果一个线程已经获取了锁，再次尝试获取锁时会失败，此时线程会进入等待状态。但是如果在等待期间，持有锁的线程又尝试获取锁，则会导致可重入性问题。
  2. 死锁问题：如果持有锁的线程异常退出或者业务执行过长时间不释放锁，那么其他线程就会一直等待该锁，从而导致死锁问题。
  3. 定时任务续租问题：虽然定时任务可以续租锁，但是无法保证定时任务一定能够执行成功。如果定时任务执行失败，那么就会出现锁过期但没有自动释放的情况。

  4. 解锁问题：当线程在 finally 块中释放锁时，首先需要判断当前线程是否持有该锁。但是如果线程在业务执行期间被重新创建并获取了同一把锁，那么该判断就会失效，从而导致无法正确释放锁的问题 。

    针对上述问题的解决方案 ，下篇见。

发表评论

表情：
评论列表 （有 0 条评论，122人围观）

还没有评论，来说两句吧...

相关阅读