Java 利用 redis 实现分布式锁

叁歲伎倆 2023-03-13 02:15 5阅读 0赞

在现在编程语言中，接触过多线程的人多多少少都对锁有一定的了解。简单来说，多线程中的锁就是在多线程运行的环境下，多个线程共享同一个资源，当对资源进行变更的时候，能保证资源的一致性机制。在分布式环境下，原来简单的多线程锁就不管用了，也就是需要分布式锁来保证多个服务共享的资源的一致性。

接下来就简单讨论下基于java通过redis实现分布式锁，实现分布式锁需要满足以下的要求：

*  支持立即获取锁方式，如果获取到返回true，获取不到返回false
 *  支持等待获取锁方式，如果获取到，直接返回true，获取不到等待一段时间，在这段时间内重复尝试获取，如果尝试获取成功，返回true，等待时间过后还获取不到，返回false
 *  不能产生死锁的情况
 *  不能释放非自己加的锁

## 加锁 ##

通过redis来实现分布式锁的加锁逻辑如下图所示

Created with Raphaël 2.2.0生成锁定的 key生成锁定的 valuekey 是否存在未获得 key 的锁保存 key-value设置 key 的过期时间获得 key 的锁yesno

根据以上逻辑，实现上锁的核心代码如下：

key = KEY_PRE + key;
       String value = this.fetchLockValue();
       if (SET_SUCCESS.equals(jedis.set(key, value, "NX", "EX", 5000))) { 
           return value;
       }

要在分布式环境中正确的实现加锁操作，“判断 key 是否存在”、“保存 key-value”、“设置key过期时间”这三个操作必须是原子操作，如果不是原子操作，则可能会出现以下两种情况：

*  在 “判断 key 是否存在” 得出key不存在的结果步骤后，“保存 key-value” 步骤前，另一个客户端执行同样的逻辑，并且执行到了 “判断 key 是否存在”步骤， 同样得出了 key 不存在的结果。这样会导致多个客户端获得到了同一把锁；
 *  在客户端执行完 “保存 key-value” 步骤后，需要设置一个 key 的过期时间，以防止客户端因为代码质量未解锁，再或者进程崩溃等情况未解锁导致的死锁情况。在 “保存 key-value” 步骤之后，“设置 key 的过期时间” 步骤之前，可能进程崩溃，导致 “设置 key 的过期时间” 步骤失败；

## 解锁 ##

解锁的基本流程如下：

Created with Raphaël 2.2.0key 是否存在判断是否自己持有锁删除 key-value解锁成功解锁失败yesnoyesno

根据以上逻辑，在代码中解锁的核心代码如下：

key = KEY_PRE + key;
       String command = "if redis.call('get',KEYS[1])==ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end";
       if (RELEASE_SUCCESS.equals(jedis.eval(command, Collections.singletonList(key), Collections.singletonList(value)))) { 
           return true;
       }

解锁和加锁的时候一样，“key 是否存在”、“判断是否自己是否持有锁”、“删除 key-value” 这三步操作需要是原子操作，否则当一个客户端执行完 “判断是否自己持有锁” 步骤后，得出自己有锁的结论，此时锁的过期时间到了，自动被redis释放了，同时另一个客户端有基于这个 key 加锁成功，如果第一个客户端还继续执行 “删除 key-value” 步骤，就会将不属于自己的锁给释放了。  
在这里我们利用以上代码中 redis 执行 Lua 脚本的能力来解决原子操作的问题。  
另外，判断是否自己持有锁的机制是用加锁的时候的 key-value 来判断当前的 key 的值是否等于自己持有锁时获得的值。所以加锁的时候的 value 必须是一个全局唯一的字符串。

完整的实现代码如下：

public class LockRedisUtil { 
    
        private static final Logger logger = LoggerFactory.getLogger(LockRedisUtil.class);
    
        private static String SET_SUCCESS = "OK";
    
        private static String KEY_PRE = "REDIS_LOCK_";
    
        private static Long LOCK_EXPIRSE_TIME = 5000L;
    
        private static Long TRY_EXPIRSE_TIME = 3000L;
    
        private DateFormat df = new SimpleDateFormat("yyyyMMddHHmmssSSS");
    
        public String lock(String key) { 
            Jedis jedis = null;
            try { 
                jedis = new Jedis();
                key = KEY_PRE + key;
                String value = this.fetchLockValue();
                if (SET_SUCCESS.equals(jedis.set(key, value, "NX", "EX", LOCK_EXPIRSE_TIME))) { 
                    return value;
                }
            } catch (Exception e) { 
                logger.info("{}", e);
            } finally { 
                jedis.close();
            }
            return null;
        }
    
        public String tryLock(String key) { 
            Jedis jedis = null;
            try { 
                jedis = new Jedis();
                key = KEY_PRE + key;
                String value = this.fetchLockValue();
                Long firstTryTime = System.currentTimeMillis();
                // 如果没获取到锁，则在 TRY_EXPIRSE_TIME 这短时间内 每过100毫秒 尝试获取一次
                do { 
                    if (SET_SUCCESS.equals(jedis.set(key, value, "NX", "EX", LOCK_EXPIRSE_TIME))) { 
                        return value;
                    }
                    TimeUnit.MILLISECONDS.sleep(100);
                } while ((System.currentTimeMillis() - TRY_EXPIRSE_TIME) < firstTryTime);
            } catch (Exception e) { 
                logger.info("{}", e);
            } finally { 
                jedis.close();
            }
            return null;
        }
    
        public boolean unLock(String key, String value) { 
            Long RELEASE_SUCCESS = 1L;
            Jedis jedis = null;
            try { 
                jedis = new Jedis();
                key = KEY_PRE + key;
                String command = "if redis.call('get',KEYS[1])==ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end";
                if (RELEASE_SUCCESS.equals(jedis.eval(command, Collections.singletonList(key), Collections.singletonList(value)))) { 
                    return true;
                }
            } catch (Exception e) { 
                logger.info("{}", e);
            } finally { 
                jedis.close();
            }
            return false;
        }
    
        /** * 生成加锁的唯一字符串 * * @return 唯一字符串 */
        private String fetchLockValue() { 
            return UUID.randomUUID().toString() + "_" + df.format(new Date());
        }
    }

## 测试代码 ##

通过创建一个线程池来模拟并发场景下的调用变更共享资源的状况，其中 getNumUseLock 方法里使用了分布式锁，而 getNumNoLock 方法则是正常的处理，未使用相关锁机制，方法里获取数据的方式是从 redis 里获取一个数，然后给这个数自加 1 ，再存回 redis 。

public class ThreadDemoA { 
    
        private static int POOL_NUM = 10;
    
        public static void main(String[] args) { 
            HandleData handleData = new HandleData();
            handleData.delNum();
            ExecutorService executorService = newFixedThreadPool(5);
            List<RunnableThread> threads = Lists.newArrayList();
            for (int i = 0; i < POOL_NUM; i++) { 
                RunnableThread thread = new RunnableThread("a" + i);
                threads.add(thread);
            }
            threads.forEach(item -> executorService.execute(item));
            executorService.shutdown();
        }
    
        static class RunnableThread implements Runnable { 
    
            private String name;
    
            private HandleData handleData = new HandleData();
    
            public RunnableThread(String name) { 
                this.name = name;
            }
    
            @Override
            public void run() { 
                int num = handleData.getNumUseLock();    // 使用了分布式锁处理
    // int num = handleData.getNumNoLock(); // 未使用相关锁机制
                System.out.println("thread " +  name + " : " + num);
            }
        }
    
        static class HandleData { 
    
            private final String KEY = "demo_key_01";
    
            private final String LOCK_KEY = "20200509";
    
            RedisUtilNonConfig redisUtil = new RedisUtilNonConfig();
    
            LockRedisUtil lockRedisUtil = new LockRedisUtil();
    
            public Integer getNumUseLock() { 
                String value = lockRedisUtil.lock(LOCK_KEY);
                if (value == null) { 
                    value = lockRedisUtil.tryLock(LOCK_KEY);
                }
                Integer num = this.getNumNoLock();
                lockRedisUtil.unLock(LOCK_KEY, value);
                return num;
            }
    
            public Integer getNumNoLock() { 
                if (!redisUtil.hasKey(KEY)) { 
                    redisUtil.setValue(KEY, "1");
                } else { 
                    Integer num = Integer.valueOf(redisUtil.getValue(KEY));
                    num++;
                    redisUtil.setValue(KEY, num.toString());
                }
                return Integer.valueOf(redisUtil.getValue(KEY));
            }
    
            public Boolean delNum() { 
                redisUtil.delValue(KEY);
                return true;
            }
        }
    }

## 测试结果 ##

使用了分布式锁处理的结果输出如下：

thread a4 : 1
    thread a3 : 2
    thread a6 : 3
    thread a7 : 4
    thread a8 : 5
    thread a9 : 6
    thread a1 : 7
    thread a0 : 8
    thread a5 : 9
    thread a2 : 10

未使用相关锁机制处理的结果输出如下：

thread a0 : 1
    thread a5 : 2
    thread a1 : 3
    thread a6 : 3
    thread a8 : 4
    thread a7 : 4
    thread a9 : 5
    thread a3 : 5
    thread a2 : 6
    thread a4 : 6

从测试结果可以看到，使用分布式锁可以有效的避免多线程环境下对同一个资源进行调用变更无法同步的问题，即在分布式环境下，多个服务对同一个资源进行处理变更，就可以使用类似的分布式锁来锁定当前资源，只有获取到锁的服务才可以进行相关处理变更资源，其它服务只能等待重新尝试。