基于 Redis 实现分布式锁，看完就懂了

待我称王封你为后i 2022-11-21 14:49 234阅读 0赞

### 基于 Redis 实现分布式锁，看完就懂了 ###

*  思考
 *  初版
 *  解决分布式线程安全问题
 *  解决死锁
 *  优化改造，阻塞式分布式锁
 *  优化改造，线程可重入
 *  优化改造，自己的锁仅有自己能解锁
 *  写个测试类测试下
 *  完整代码

# 思考 #

设计一个分布式锁需要考虑哪些因素？

1.  分布式锁本身线程安全
2.  假如出现服务宕机，是否会出现死锁情况
3.  阻塞式分布式锁
4.  线程重入问题
5.  自己的锁只有自己可以释放

# 初版 #

先来个最简单版本，自然上述问题都没有做考虑  
其中我认为最为严重的问题就是锁本身就存在线程安全问题

假如同时进行两个线程 A 与 B  
假如线程 A 在执行 `redisTemplate.opsForValue().set(lockKey, "lock")` 时候卡住了，比如网络波动  
这时线程 B 执行了 `if (!redisTemplate.hasKey(lockKey))` 就也通过了判断，导致两个线程都成功获取到锁

/** * Redis 分布式锁 * @author: linjinp * @create: 2020-07-27 10:58 **/
    @Slf4j
    public class RedisLock { 
    
    	@Resource
        private RedisTemplate<Serializable, Object> redisTemplate;
        
        // 锁的 Key
        @Setter
        private final String lockKey;
        
    	public RedisLock(String lockKey) { 
            this.lockKey = lockKey;
        }
        
    	/** * 获取锁 * @return */
        public boolean  lock() { 
        	boolean success = false;
        	// 判断 key 是否存在
            if (!redisTemplate.hasKey(lockKey)) { 
                redisTemplate.opsForValue().set(lockKey, "lock");
                success = true;
            }
            return success;
        }
        
    	/** * 释放锁 */
        public void unLock() { 
        	redisTemplate.delete(lockKey);
        }
    }

# 解决分布式线程安全问题 #

会存在线程安全的根本原因是因为：`SET IF NOT EXIST` 这个操作本身是非线程安全的，而 Redis 是有提供线程安全的操作的。

Redis 的 Setnx 操作，会判断 key 是否已经存在，如果存在返回 `false`，否则的话就会 Set 成功，并且返回 `true`

/** * Redis 分布式锁 * @author: linjinp * @create: 2020-07-27 10:58 **/
    @Slf4j
    public class RedisLock { 
    
    	@Resource
        private RedisTemplate<Serializable, Object> redisTemplate;
        
        // 锁的 Key
        @Setter
        private final String lockKey;
        
    	public RedisLock(String lockKey) { 
            this.lockKey = lockKey;
        }
        
    	/** * 获取锁 * @return */
        public boolean  lock() { 
        	// key 已存在，返回 false
        	// key 不存在，设置成功并返回 true
            return redisTemplate.opsForValue().setIfAbsent(lockKey, "lock");
        }
        
    	/** * 释放锁 */
        public void unLock() { 
        	redisTemplate.delete(lockKey);
        }
    }

# 解决死锁 #

上面的方案依然存在问题

假如一个线程获取到锁后，在执行 unlock 之前服务宕机了，这下好了下次再来获取锁将永远死锁，为了解决这个问题，利用 Redis 的超时机制来保证锁不至于永远存在

/** * Redis 分布式锁 * @author: linjinp * @create: 2020-07-27 10:58 **/
    @Slf4j
    public class RedisLock { 
    
    	@Resource
        private RedisTemplate<Serializable, Object> redisTemplate;
        
    	// 锁超时时间默认值 30s
        private final static long DEFAULT_EXPIRE = 30000;
        
        // 锁的 Key
        @Setter
        private final String lockKey;
        
    	public RedisLock(String lockKey) { 
            this.lockKey = lockKey;
        }
        
    	/** * 获取锁 * @return */
        public boolean lock() { 
        	// key 已存在，返回 false
        	// key 不存在，设置成功并返回 true
        	// 30s 超时
            return redisTemplate.opsForValue().setIfAbsent(lockKey, "lock", DEFAULT_EXPIRE, TimeUnit.MILLISECONDS);
        }
        
    	/** * 释放锁 */
        public void unLock() { 
        	redisTemplate.delete(lockKey);
        }
    }

# 优化改造，阻塞式分布式锁 #

上述的非阻塞式分布式锁，一些大问题算是解决了，属于基本能用的状态

但是每次获取锁我都要判断下锁是否获取成功，非常的不好用，很不友好，重复劳动力压榨

因此我们改造下，改造成阻塞式分布式锁

每次获取锁假如已经已经有线程持有锁了，我们就进行等待，等到那个线程锁释放了再获取

/** * Redis 分布式锁 * @author: linjinp * @create: 2020-07-27 10:58 **/
    @Slf4j
    public class RedisLock { 
    
    	@Resource
        private RedisTemplate<Serializable, Object> redisTemplate;
        
    	// 锁超时时间默认值 30s
        private final static long DEFAULT_EXPIRE = 30000;
        
        // 锁阻塞时，线程睡眠时间
        private final static long LOCK_WAITING = 3000;
        
        // 锁的 Key
        @Setter
        private final String lockKey;
        
    	public RedisLock(String lockKey) { 
            this.lockKey = lockKey;
        }
        
    	/** * 获取锁 * @return */
        public void lock() throws InterruptedException { 
        	// 如果已经有线程持有锁了，就先睡眠 3 秒，再次尝试获取锁
        	while (!redisTemplate.opsForValue().setIfAbsent(lockKey, "lock", DEFAULT_EXPIRE, TimeUnit.MILLISECONDS)) { 
                Thread.sleep(LOCK_WAITING);
            }
        }
        
    	/** * 释放锁 */
        public void unLock() { 
        	redisTemplate.delete(lockKey);
        }
    }

# 优化改造，线程可重入 #

上面的分布式锁已经在绝大多数场景下都可以使用了，但依然存在不适用的场景

比如线程 A 获取了锁成功后，在未释放的情况下再次获取锁，就不行了

try { 
    	lock.lock();
    	...
    	lock.lock();
    	...
    } finally { 
    	lock.unlock();
    	lock.unlock();
    }

利用 ThreadLocal 解决分布式锁重入问题，并且做到重入多少次，就要解锁多少次

可能会有疑问，难道不需要把线程 ID 存到 Redis 中，判断下线程是否重复吗？  
答：不需要，想想也知道线程的 ID 是不会重复的（自定义除外），会有这种想法是因为一种固化思维

ThreadLocal 的作用是什么？  
答：ThreadLocal 中的变量仅有当前线程能看到，用于做全局变量的线程隔离

/** * Redis 分布式锁 * @author: linjinp * @create: 2020-07-27 10:58 **/
    @Slf4j
    public class RedisLock { 
    
    	@Resource
        private RedisTemplate<Serializable, Object> redisTemplate;
    
        // 存储当前线程的唯一 UUID
        ThreadLocal<String> threadUUID = new ThreadLocal<>();
    
        // 当前线程的重入次数
        ThreadLocal<Integer> threadReentryTimes = ThreadLocal.withInitial(() -> 0);
        
    	// 锁超时时间默认值 30s
        private final static long DEFAULT_EXPIRE = 30000;
        
        // 锁阻塞时，线程睡眠时间
        private final static long LOCK_WAITING = 3000;
        
        // 锁的 Key
        @Setter
        private final String lockKey;
        
    	public RedisLock(String lockKey) { 
            this.lockKey = lockKey;
        }
        
    	/** * 获取锁 * @return */
        public void lock() throws InterruptedException { 
        	// 唯一 UUID
            String uuid = IdUtil.randomUUID();
    
            // 如果不存在则新增，存在不改变原有值
            // 如果阻塞，则线程睡眠
            // 场景1：如果发现 threadUUID 为空，说明非当前线程，尝试获取锁
            if (threadUUID.get() == null) { 
                while (!redisTemplate.opsForValue().setIfAbsent(lockKey, uuid, DEFAULT_EXPIRE, TimeUnit.MILLISECONDS)) { 
                    Thread.sleep(LOCK_WAITING);
                }
                threadUUID.set(uuid);
                log.info("[{}]加锁成功", lockKey);
            } else { 
                // 场景2：如果发现 threadUUID 不为空，说明为当前线程重入，则将重入次数 +1
                threadReentryTimes.set(threadReentryTimes.get() + 1);
                log.info("[{}]线程重入成功，重入次数：{}", lockKey, threadReentryTimes.get());
            }
        }
        
    	/** * 释放锁 */
        public void unLock() { 
        	if (threadReentryTimes.get() > 0) { 
                threadReentryTimes.set(threadReentryTimes.get() - 1);
                log.info("[{}]线程重入解锁成功，剩余重入次数：{}", lockKey, threadReentryTimes.get());
            } else { 
                // 场景3：如果 threadUUID 不为空，threadReentryTimes 小于等于 0，Redis 解锁，正常不可能小于 0
                if (threadUUID.get().equals(redisTemplate.opsForValue().get(lockKey))) { 
                    redisTemplate.delete(lockKey);
                    log.info("[{}]解锁成功", lockKey);
                    // 防止 ThreadLocal 内存泄漏
                    threadReentryTimes.remove();
                    threadUUID.remove();
                }
            }
        }
    }

# 优化改造，自己的锁仅有自己能解锁 #

这个是个非常容易遗漏的问题，即自己的锁只有自己能解锁，虽然一般只要代码不瞎写就不会存在这种问题，但是还是应该考虑下

/** * Redis 分布式锁 * @author: linjinp * @create: 2020-07-27 10:58 **/
    @Slf4j
    public class RedisLock { 
    
    	@Resource
        private RedisTemplate<Serializable, Object> redisTemplate;
    
        // 存储当前线程的唯一 UUID
        ThreadLocal<String> threadUUID = new ThreadLocal<>();
    
        // 当前线程的重入次数
        ThreadLocal<Integer> threadReentryTimes = ThreadLocal.withInitial(() -> 0);
        
    	// 锁超时时间默认值 30s
        private final static long DEFAULT_EXPIRE = 30000;
        
        // 锁阻塞时，线程睡眠时间
        private final static long LOCK_WAITING = 3000;
        
        // 锁的 Key
        @Setter
        private final String lockKey;
        
    	public RedisLock(String lockKey) { 
            this.lockKey = lockKey;
        }
        
    	/** * 获取锁 * @return */
        public void lock() throws InterruptedException { 
        	// 唯一 UUID
            String uuid = IdUtil.randomUUID();
    
            // 如果不存在则新增，存在不改变原有值
            // 如果阻塞，则线程睡眠
            // 场景1：如果发现 threadUUID 为空，说明非当前线程，尝试获取锁
            if (threadUUID.get() == null) { 
                while (!redisTemplate.opsForValue().setIfAbsent(lockKey, uuid, DEFAULT_EXPIRE, TimeUnit.MILLISECONDS)) { 
                    Thread.sleep(LOCK_WAITING);
                }
                threadUUID.set(uuid);
                log.info("[{}]加锁成功", lockKey);
            } else { 
                // 场景2：如果发现 threadUUID 不为空，说明为当前线程重入，则将重入次数 +1
                threadReentryTimes.set(threadReentryTimes.get() + 1);
                log.info("[{}]线程重入成功，重入次数：{}", lockKey, threadReentryTimes.get());
            }
        }
        
    	/** * 释放锁 */
        public void unLock() { 
        	// 如果 key 对应的 value 与当前线程的 UUID 一致，说明是当前线程来解锁，才可解锁
            // 场景1：如果 threadUUID 空的，说明就没加过锁，直接跳过
            if (StringUtils.isNotBlank(threadUUID.get())) { 
                // 场景2：如果 threadUUID 不为空，threadReentryTimes 大于 0，说明重入线程解锁，重入数量 -1
                if (threadReentryTimes.get() > 0) { 
                    threadReentryTimes.set(threadReentryTimes.get() - 1);
                    log.info("[{}]线程重入解锁成功，剩余重入次数：{}", lockKey, threadReentryTimes.get());
                } else { 
                    // 场景3：如果 threadUUID 不为空，threadReentryTimes 小于等于 0，Redis 解锁，正常不可能小于 0
                    if (threadUUID.get().equals(redisTemplate.opsForValue().get(lockKey))) { 
                        redisTemplate.delete(lockKey);
                        log.info("[{}]解锁成功", lockKey);
                        // 防止 ThreadLocal 内存泄漏
                        threadReentryTimes.remove();
                        threadUUID.remove();
                    }
                }
            }
        }
    }

# 写个测试类测试下 #

`final：` 这个关键字是为了保证锁的引用不可变，防止被篡改，同时保证线程间的可见性

/** * Redis 分布式锁测试 * * @Author: linjinp * @Date: 2020/10/26 11:55 */
    @RunWith(SpringRunner.class)
    @SpringBootTest
    public class RedisLockTest { 
    
        private final static RedisLock lock = new RedisLock("test");
    
        /** * Redis 分布式锁测试 * @return */
        @Test
        public void redisLockTest() throws InterruptedException { 
            try { 
                lock.lock();
                lock.lock();
                lock.lock();
            } finally { 
                lock.unLock();
                lock.unLock();
                lock.unLock();
            }
        }
    }

![在这里插入图片描述][20201102114232841.png_pic_center]

# 完整代码 #

/** * Redis 分布式锁 * @author: linjinp * @create: 2020-07-27 10:58 **/
    @Slf4j
    public class RedisLock { 
    
    	@Resource
        private RedisTemplate<Serializable, Object> redisTemplate;
    
        // 存储当前线程的唯一 UUID
        ThreadLocal<String> threadUUID = new ThreadLocal<>();
    
        // 当前线程的重入次数
        ThreadLocal<Integer> threadReentryTimes = ThreadLocal.withInitial(() -> 0);
    
        // 锁超时时间默认值 30s
        private final static long DEFAULT_EXPIRE = 30000;
    
        // 锁阻塞时，线程睡眠时间
        private final static long LOCK_WAITING = 3000;
    
        // 锁的 Key
        @Setter
        private final String lockKey;
    
        // 超时时间
        @Setter
        private final long expire;
    
        public RedisLock(String lockKey) { 
            this.lockKey = lockKey;
            this.expire = DEFAULT_EXPIRE;
        }
    
        public RedisLock(String lockKey, long expire) { 
            this.lockKey = lockKey;
            this.expire = expire;
        }
    
        /** * 获取锁 * @return */
        public void lock() throws InterruptedException { 
            lock(expire);
        }
    
        /** * 获取锁 * @param millisecond * @return */
        public void lock(long millisecond) throws InterruptedException { 
            // 唯一 UUID
            String uuid = IdUtil.randomUUID();
    
            // 如果不存在则新增，存在不改变原有值
            // 如果阻塞，则线程睡眠
            // 场景1：如果发现 threadUUID 为空，说明非当前线程，尝试获取锁
            if (threadUUID.get() == null) { 
                while (!redisTemplate.opsForValue().setIfAbsent(lockKey, uuid, millisecond, TimeUnit.MILLISECONDS)) { 
                    Thread.sleep(LOCK_WAITING);
                }
                threadUUID.set(uuid);
                log.info("[{}]加锁成功", lockKey);
            } else { 
                // 场景2：如果发现 threadUUID 不为空，说明为当前线程重入，则将重入次数 +1
                threadReentryTimes.set(threadReentryTimes.get() + 1);
                log.info("[{}]线程重入成功，重入次数：{}", lockKey, threadReentryTimes.get());
            }
        }
    
        /** * 释放锁 */
        public void unLock() { 
            // 如果 key 对应的 value 与当前线程的 UUID 一致，说明是当前线程来解锁，才可解锁
            // 场景1：如果 threadUUID 空的，说明就没加过锁，直接跳过
            if (StringUtils.isNotBlank(threadUUID.get())) { 
                // 场景2：如果 threadUUID 不为空，threadReentryTimes 大于 0，说明重入线程解锁，重入数量 -1
                if (threadReentryTimes.get() > 0) { 
                    threadReentryTimes.set(threadReentryTimes.get() - 1);
                    log.info("[{}]线程重入解锁成功，剩余重入次数：{}", lockKey, threadReentryTimes.get());
                } else { 
                    // 场景3：如果 threadUUID 不为空，threadReentryTimes 小于等于 0，Redis 解锁，正常不可能小于 0
                    if (threadUUID.get().equals(redisTemplate.opsForValue().get(lockKey))) { 
                        redisTemplate.delete(lockKey);
                        log.info("[{}]解锁成功", lockKey);
                        // 防止 ThreadLocal 内存泄漏
                        threadReentryTimes.remove();
                        threadUUID.remove();
                    }
                }
            }
        }
    }

[20201102114232841.png_pic_center]: /images/20221120/0dd0a88bb191405086544925d77b1266.png