Redis实现分布式锁的正确方法

共计 1531 个字符，预计需要花费 4 分钟才能阅读完成。

这里主要记录项目中使用基于 Redis 的分布式锁所遇到的问题及解决方案；

业务场景

我的业务场景是这样的，我们服务有库存模块，而我的服务又是多节点部署，要高峰期会存在库存差异，后面分析问题之后，打算采用 redis 实现分布式锁（主要的原因是服务已经集成了 redis，不需要做额外的配置）

踩坑 1. 数据库事务超时

不要感觉奇怪，分布式锁怎么会导致数据库事务超时呢?
我的代码大概是这样的：

伪代码
@Transaction(readOnly=false)
void update(){do{redis=JedisUtil.getJedis();
        flag = getLock(key,redis);
        if(flag){update();
        }
    }while(true)
}

当你的 key 长时间获取不到锁，并且数据库事务都有超时时间的限制，那么就会出现数据库事务超时问题；
解决方案

数据库事务改为手动提交事务；

踩坑 2. redis key 过期，而业务没有执行完

我的 key 的过期时间设置的是 30s，如果 30 秒业务还没有执行完毕，锁就会自动释放，锁释放之后，其它线程又会去占用锁，同样会导致问题的发生；
解决方案

最简单的解决方案就是使用 redisson；
如果非要用 redis 来解决的话，只能使用定时器去检测 key，如果说 key 还有 2 秒就快过期了，那么再为 key 重新设置 30 秒的过期时间；

踩坑 3. redis 连接池爆满

分布式锁刚加上之后，生产出现一个问题，就是：redis.clients.jedis.exceptions.JedisConnectionException: Could not get a resource from the pool
解决办法
开始查代码，发现是开发人员没有对连接进行释放；

修复 bug 之后，又在线上跑了一段时间，又出现了 redis.clients.jedis.exceptions.JedisConnectionException: Could not get a resource from the pool
解决办法

void update(){do{redis=JedisUtil.getJedis();
        flag = getLock(key,redis);
        if(flag){update();
        }else{// 释放当前 redis 连接
            // 由于我们的业务场景属于比较耗时的业务型，所以在这里休眠 1000 毫秒
            redis.close();
            sleep(1000);
        }
    }while(true)
}

1. 当前请求获取锁，如果获取不到，则释放当前连接，并休眠一会；
2. 合理配置 redis 连接池大小，主要参考具体业务场景的并发量来设置；

踩坑 4. 解铃还须系铃人

回顾一下加锁的参数：

set(key, vlue,"NX","PX", 30000);

其中：value，我使用它来表示加锁人，必须是一个唯一的标识

比如：
A 线程 key=test value=01
B 线程 key=test value=02

如果 A 线程执行业务耗时超过了锁的持有时间，锁会自动释放；锁自动释放之后，线程 B 又加锁成功，但是，此时 A 线程执行完业务逻辑之后，去释放锁，但 A 线程的锁已经自动释放了，如果没有 value 来标识的话，它可能就会去释放 B 线程的锁；

踩坑 5. redis 集群实现分布式锁

这种情况我没有遇到，因为公司的 redis 集群做了改进；

先说一下这种问题产生的原因：
如果 master 节点由于某原因发生了主从切换，那么就会出现锁丢失的情况；

• 在 master 节点上拿到了锁；
• 但是这个加锁的 key 还没有同步到 slave 节点；
• master 故障，发生故障转移，slave 节点升级为 master 节点；
• 故导致锁丢失；

解决办法

需要通过使用 redlock 算法；
或使用 redisson，它有对 redlock 算法做封装；

：