分布式锁实现的几种方式(DB,Redis,Zookeeper)

首先，为了确保分布式锁可用，我们至少要确保锁的实现同时满足以下四个条件：

1.互斥性。在任意时刻，只有一个客户端能持有锁。
2.不会发生死锁。即使有一个客户端在持有锁的期间崩溃而没有主动解锁，也能保证后续其他客户端能加锁。
3.具有容错性。只要大部分的Redis节点正常运行，客户端就可以加锁和解锁。
4.解铃还须系铃人。加锁和解锁必须是同一个客户端，客户端自己不能把别人加的锁给解了。

应用场景：订单模块新增一条订单，商品模块减少库存，共用一个数据库

不加锁代码：

/*** 减少库存-不加锁-会存在超卖问题** @param productId* @return*/@Transactional@Overridepublic Pair<Boolean, String> reductStock(Integer productId) {Integer stock = this.productMapper.getStock(productId);if (stock <= 0) {return Pair.of(false, MessageFormat.format("【线程" + Thread.currentThread().getId() + "】库存不足,剩余库存{0}", stock));}try {Thread.sleep(5000);} catch (InterruptedException e) {e.printStackTrace();}this.productMapper.reductStock(productId);return Pair.of(true, MessageFormat.format("【线程" + Thread.currentThread().getId() + "】减少库存成功,剩余库存{0}", this.productMapper.selectByPrimaryKey(productId).getStock()));}

<select id="getStock" resultType="java.lang.Integer">select stock from product where id=#{productId}
</select>

<update id="reductStock">update product set stock=stock-1 where id=#{productId}
</update>

方式一（数据库-for update悲观锁）：

/**
* 总结几个点：
* 1.for update 锁必须使用事务，不然起不到真正锁的效果，数据库默认的写语句都是执行即提交
* 2.for update 锁是行锁，查询语句必须走主键索引，否则可能锁掉其他数据或者整个表
* 3.for update加锁后，其他线程或进程中数据库客户端执行语句和java客户端执行语句都会进入等待状态
* 4.for update加锁后，还没来得及解锁程序异常时，事务会回滚，释放锁
*/

/*** 减少库存-数据库for update锁-悲观锁** @param productId* @return*/@Transactional@Overridepublic Pair<Boolean, String> reductStockByLock(Integer productId) {//加锁Integer stock = this.productMapper.getStockByLock(productId);logger.info("【线程" + Thread.currentThread().getId() + "】加锁成功。。。");if (stock <= 0) {return Pair.of(false, MessageFormat.format("【线程" + Thread.currentThread().getId() + "】库存不足,剩余库存{0}", stock));}try {Thread.sleep(5000);} catch (InterruptedException e) {e.printStackTrace();}//如果加锁后程序遇到异常，事务会释放当前锁资源//Integer num = 1 / 0;//解锁this.productMapper.reductStock(productId);logger.info("【线程" + Thread.currentThread().getId() + "】解锁成功。。。");return Pair.of(true, MessageFormat.format("【线程" + Thread.currentThread().getId() + "】减少库存成功,剩余库存{0}", this.productMapper.selectByPrimaryKey(productId).getStock()));}

<select id="getStockByLock" resultType="java.lang.Integer">select stock from product where id=#{productId} for update
</select>

方式一（数据库-乐观锁）：

/*** 减少库存-数据库version锁-乐观锁** @param productId* @return*/@Transactional@Overridepublic Pair<Boolean, String> reductStockByVersionLock(Integer productId) {Product product = this.productMapper.selectByPrimaryKey(productId);Integer stock = product.getStock();String version = product.getLockVersion();if (product.getStock() <= 0) {return Pair.of(false, MessageFormat.format("【线程" + Thread.currentThread().getId() + "】,版本:" + version + ",库存不足,剩余库存{0}", stock));}try {Thread.sleep(5000);} catch (InterruptedException e) {e.printStackTrace();}//更新的时候条件where要带上versionInteger result = this.productMapper.reductStockByVersionLock(productId, version);if (null == result || result <= 0) {return Pair.of(true, MessageFormat.format("【线程" + Thread.currentThread().getId() + "】,版本:" + version + ",减少库存失败,剩余库存{0}", this.productMapper.selectByPrimaryKey(productId).getStock()));}return Pair.of(true, MessageFormat.format("【线程" + Thread.currentThread().getId() + "】,版本:" + version + ",减少库存成功,剩余库存{0}", this.productMapper.selectByPrimaryKey(productId).getStock()));}

<update id="reductStockByVersionLock">update product set stock=stock-1,lock_version=lock_version+1 where id=#{productId} and lock_version=#{version}
</update>

方式二（Redis-Lua实现）：

错误版本：

/*** 加锁** @param lockKeyEnum 锁类型* @param lockDataId  锁对应的业务表数据id*/public Boolean lock(LockKeyEnum lockKeyEnum, String lockDataId) {Jedis jedis = null;String lockKey = lockKeyEnum.getCode() + "_" + lockDataId;try {jedis = jedisPool.getResource();if (null == jedis) {return false;}//1表示成功 0表示失败（该key已存在）if (jedis.setnx(lockKey, lockDataId).equals(0)) {return false;}jedis.expire(lockKey, 10);return true;} catch (Exception e) {logger.error("加锁异常。。。", e);//手动解锁this.unLock(jedis, lockKeyEnum, lockDataId);} finally {//释放资源jedis.close();}return false;}/*** 解锁** @param lockKeyEnum 锁类型* @param lockDataId  锁对应的业务表数据id*/public Boolean unLock(Jedis jedis, LockKeyEnum lockKeyEnum, String lockDataId) {jedis.del(lockKeyEnum.getCode() + "_" + lockDataId);return true;}

错误点1（造成死锁）：

加锁时：

如果执行到jedis.setnx整个程序挂掉了，这里指的是程序直接退出，try catch没有用，会导致死锁

改进1：

这里让判断赋值和设置过期时间一起执行，原子性操作

错误点2（删除掉别人的锁）：

解锁时：

假设这里有ABC三个客户端请求排队，这时A客户端抢到了锁设置过期时间为10秒钟，如果这里执行的时间比较长，当第11秒钟的时候，（客户端A的）锁过期，

客户端B马上抢到锁并上锁，这时客户端A恰好执行完成，进行解锁操作，就会把（客户端B的）锁删除，这时客户端B处于裸奔状态，客户端C马上抢到锁并上锁，

这时问题来了，客户端B和客户端C的操作不是原子性，会出现互相抢资源，出现超卖的情况出现

改进2

给每一次客户端请求都加上version版本标记，解锁的时候判断是不是自己客户端请求的锁

加锁时：

解锁时：

错误点3（非原子操作删除掉别人的锁）：

改进了2次后，似乎没问题了，但其实还是有一个隐藏的问题，看这段解锁代码，

假设客户端A加锁后，正常执行完成，在解锁的时候，判断版本这一句代码执行完的时候，（客户端A的）锁正好过期，然后（客户端B的）锁加上，

客户端A的解锁操作继续执行，这样客户端B的锁就被删除掉，客户端B就变成裸奔状态了

改进3

为了解决解锁时不是原子性操作的问题，解锁时必须引入lua语法来保证该操作是原子性的操作

所以最终的代码是：

package com.study.utils;import com.study.enums.LockKeyEnum;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;import java.util.Collections;
import java.util.UUID;@Component
public class RedisLockByLua {private static final Logger logger = LoggerFactory.getLogger(RedisLockByLua.class);/*** 加锁-返回状态*/private static final String LOCK_SUCCESS = "OK";/*** 解锁-返回状态*/private static final Long RELEASE_SUCCESS = 1L;/*** 设置-如果不存在就添加*/private static final String SET_IF_NOT_EXIST = "NX";/*** 设置-过期时间*/private static final String SET_WITH_EXPIRE_TIME = "PX";/*** 过期时间(10秒钟)*/private static final Long EXPIRE_TIME = 20 * 1000L;/*** 加锁** @param lockKeyEnum*         锁类型* @param lockDataId*         锁对应的业务表数据id* @param lockVersion*         锁对应的版本（UUID）*/public Boolean lock(Jedis jedis, LockKeyEnum lockKeyEnum, String lockDataId, String lockVersion) {String lockKey = lockKeyEnum.getCode() + "_" + lockDataId;try {while (true) {String result = jedis.set(lockKey, lockVersion, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, EXPIRE_TIME);if (LOCK_SUCCESS.equals(result)) {//加锁成功，跳出循环return true;}logger.info("【线程{}】【版本:{}】等待加锁。。。", Thread.currentThread().getId(), lockVersion);}} catch (Exception e) {logger.error("【线程{" + Thread.currentThread().getId() + "}】【版本:{" + lockVersion + "}】加锁异常。。。", e);//手动解锁this.unLock(jedis, lockKeyEnum, lockDataId, lockVersion);return false;}}/*** 解锁** @param lockKeyEnum*         锁类型* @param lockDataId*         锁对应的业务表数据id* @param lockVersion*         锁对应的版本*/public Boolean unLock(Jedis jedis, LockKeyEnum lockKeyEnum, String lockDataId, String lockVersion) {String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";Object result = jedis.eval(script, Collections.singletonList(lockKeyEnum.getCode() + "_" + lockDataId), Collections.singletonList(lockVersion));if (RELEASE_SUCCESS.equals(result)) {return true;}return false;}}

方式三（Redission实现）：

这里有一篇文章详解了Redisson原理机制，大概Redisson内部加锁和解锁都用了Lua实现，保证原子性操作

https://blog.csdn.net/asd051377305/article/details/108384490

 /*** 减少库存-Redis-Redisson锁** @param productId* @return*/@Overridepublic Pair<Boolean, String> reduceStockByRedisson(Integer productId) throws Exception {RLock rLock = null;try {//1.配置String address = "redis://" + REDIS_HOSTNAME + ":" + REDIS_PORT;Config config = new Config();config.useSingleServer().setAddress(address).setDatabase(0);//2.构建RedissonClientRedissonClient redissonClient = Redisson.create(config);String lockKey = LockKeyEnum.Product_Stock.getCode() + "_" + productId;rLock = redissonClient.getLock(lockKey);//等待获取锁的最大等待时间，超过这个值，则认为获取锁失败Long waitTimeOut = 30L;//锁的超时时间，超过这个时间锁会自动失效(设置的时间比业务处理时间大，保证有效期内业务能处理完成)Long leaseTime = 10L;logger.info("【线程{}】准备获取锁。。。", Thread.currentThread().getId());/*** 1.获取不到锁时会阻塞,内部实现了while(true)循环* 2.1如果我们指定了锁的超时时间 leaseTime，就发送给redis执行脚本，进行占锁，默认超时就是我们制定的时间，不会自动续期；* 2.2如果我们未指定锁的超时时间 leaseTime，就使用 lockWatchdogTimeout = 30 * 1000 【看门狗默认时间】* 只要占锁成功，就会启动一个定时任务【重新给锁设置过期时间，新的过期时间就是看门狗的默认时间】,每隔10秒都会自动再续成30秒；*///3.加锁//Boolean result = rLock.tryLock(waitTimeOut, TimeUnit.SECONDS);Boolean result = rLock.tryLock(waitTimeOut, leaseTime, TimeUnit.SECONDS);if (result) {logger.info("【线程{}】获取锁成功，开始业务。。。", Thread.currentThread().getId());Product product = this.productMapper.selectByPrimaryKey(productId);Integer stock = product.getStock();if (stock <= 0) {return Pair.of(false, MessageFormat.format("【线程{0}】库存不足,剩余库存{1}", Thread.currentThread().getId(), stock));}Thread.sleep(5000);this.productMapper.reductStock(productId);logger.info("【线程{}】业务执行完成。。。", Thread.currentThread().getId());return Pair.of(true, MessageFormat.format("【线程{0}】减少库存成功,剩余库存{1}", Thread.currentThread().getId(), this.productMapper.selectByPrimaryKey(productId).getStock()));} else {logger.info("【线程{}】获取锁失败。。。", Thread.currentThread().getId());return Pair.of(false, MessageFormat.format("【线程{0}】获取锁失败。。。", Thread.currentThread().getId()));}} catch (Exception e) {logger.error("【线程" + Thread.currentThread().getId() + "】减少库存异常。。。", e);throw e;} finally {//4.最后都要解锁(判断1.解锁状态2.解锁当前线程的锁)if (null != rLock && rLock.isLocked() && rLock.isHeldByCurrentThread()) {logger.info("【线程{}】解锁。。。", Thread.currentThread().getId());rLock.unlock();}}}

方式四（Zookeeper实现）：

package com.study.utils;import com.study.enums.LockKeyEnum;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.apache.commons.lang3.StringUtils;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;import java.io.IOException;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.stream.Collectors;public class ZookeeperLock implements Lock {public static void main(String[] args) throws IOException, KeeperException, InterruptedException {//ZookeeperLock lock = new ZookeeperLock(LockKeyEnum.Product_Stock.getCode() + "_1");//lock.lock();...执行业务//Thread.sleep(1000);//lock.unlock();}private static final Logger logger = LoggerFactory.getLogger(ZookeeperLock.class);@Value("${zookeeper.url}")private String zkUrl;/*** 锁-key*/private String lockKey;/*** zk客户端*/private ZkClient zk;/*** 客户端连接服务器超时时间(毫秒)*/private static final Integer sessionTimeOut = 30000;/*** 计数器*/private CountDownLatch latch;/*** 锁-根目录地址*/private static final String rootNodePath = "/myLock";/*** 当前锁节点*/private String nowNodePath;/*** 当前锁节点 等待释放的前一个节点*/private String waitNodePath;public ZookeeperLock(String _lockKey) {this.lockKey = _lockKey;try {zk = new ZkClient(StringUtils.isEmpty(zkUrl) ? "192.168.247.129:2181" : zkUrl);if (!zk.exists(rootNodePath)) {//创建根节点(持久节点)zk.createPersistent(rootNodePath);}} catch (Exception e) {logger.error("初始化zk客户端异常。。。", e);if (null != zk) {zk.close();}}}/*** 获取锁*/@Overridepublic void lock() {if (this.tryLock()) {logger.info("【线程{}】获取锁成功。。。", Thread.currentThread().getId());return;} else {//等待锁this.waitForLock();//递归重新尝试获取锁，一旦监听到上一个锁线程释放了锁，再次重新获取锁logger.info("【线程{}】重新尝试获取锁。。。", Thread.currentThread().getId());lock();}}/*** 等待获取锁** @return*/private Boolean waitForLock() {try {logger.info("【线程{}】等待获取锁。。。", Thread.currentThread().getId());//注册监听此节点的变化IZkDataListener iZkDataListener = new IZkDataListener() {@Overridepublic void handleDataChange(String dataPath, Object data) throws Exception {}@Overridepublic void handleDataDeleted(String dataPath) throws Exception {logger.info("【线程{}】【节点{}】监控到被释放删除。。。", Thread.currentThread().getId(), dataPath);if (null != latch) {latch.countDown();}}};//子节点加入Watcher监控zk.subscribeDataChanges(waitNodePath, iZkDataListener);if (zk.exists(waitNodePath)) {logger.info("【线程" + Thread.currentThread().getId() + "】【节点{}】阻塞，【节点{}】等待解锁。。。", nowNodePath, waitNodePath);this.latch = new CountDownLatch(1);//这里阻塞，一直等待其他线程释放锁this.latch.await();logger.info("【线程" + Thread.currentThread().getId() + "】【节点{}】阻塞解除，【节点{}】解锁。。。", nowNodePath, waitNodePath);}//子节点释放Watcher监控zk.unsubscribeDataChanges(waitNodePath, iZkDataListener);return true;} catch (Exception e) {logger.error("等待获取锁异常。。。", e);throw new RuntimeException("等待获取锁异常。。。", e);}}/*** 它表示用来尝试获取锁，如果获取成功，则返回true；如果获取失败（即锁已被其他线程获取），则返回false，也就是说，这个方法无论如何都会立即返回（在拿不到锁时不会一直在那等待）** @return*/@Overridepublic boolean tryLock() {try {String splitStr = "_";//这个判断很重要,如果第一次进来tryLock,当前线程节点未指定,创建当前节点,不判断的话就会死循环,当前线程节点永远为最新产生的+1的节点if (StringUtils.isEmpty(nowNodePath)) {nowNodePath = zk.createEphemeralSequential(rootNodePath + "/" + lockKey + splitStr, "lock");}//获取所有子节点List<String> allChildrenList = zk.getChildren(rootNodePath);//筛选出当前业务锁对应的子节点List<String> nowChildrenList = allChildrenList.stream().filter(a -> a.contains(lockKey)).collect(Collectors.toList());//排序，升序Collections.sort(nowChildrenList);//节点内容:  product_stock_1_0000000001//如果当前锁节点是最小的节点if (nowNodePath.equals(rootNodePath + "/" + nowChildrenList.get(0))) {logger.info("【线程:{}】【节点:{}】为最小节点,获取到锁", Thread.currentThread().getId(), nowNodePath);return true;}logger.info("【线程:{}】【节点:{}】不是最小节点,等待获取锁", Thread.currentThread().getId(), nowNodePath);//如果当前锁节点不是最小的节点，找到比自己小1的节点List<String> smallerNodeList = nowChildrenList.stream().filter(a -> Long.valueOf(a.substring(a.lastIndexOf(splitStr) + 1)) < Long.valueOf(nowNodePath.substring(nowNodePath.lastIndexOf(splitStr) + 1))).collect(Collectors.toList());waitNodePath = rootNodePath + "/" + smallerNodeList.get(smallerNodeList.size() - 1);} catch (Exception e) {logger.error("尝试获取锁异常。。。", e);throw new RuntimeException("尝试获取锁异常。。。", e);}return false;}/*** 这个方法在拿不到锁时会等待一定的时间，在时间期限之内如果还拿不到锁，就返回false，同时可以响应中断。如果一开始拿到锁或者在等待期间内拿到了锁，则返回true** @param time* @param unit* @return* @throws InterruptedException*/@Overridepublic boolean tryLock(long time, TimeUnit unit) throws InterruptedException {return false;}/*** 解锁*/@Overridepublic void unlock() {try {//删除当前节点if (zk.exists(nowNodePath)) {zk.delete(nowNodePath);}nowNodePath = null;//释放zk客户端连接zk.close();logger.info("【线程{}】解锁成功。。。", Thread.currentThread().getId());} catch (Exception e) {logger.error("解锁异常。。。", e);throw new RuntimeException("解锁异常。。。", e);}}/*** 返回绑定到此 Lock 实例的新 Condition 实例** @return*/@Overridepublic Condition newCondition() {return null;}/*** 如果当前线程未被中断，则获取锁，可以响应中断** @throws InterruptedException*/@Overridepublic void lockInterruptibly() throws InterruptedException {}}

简单总结：

从理解的难易程度角度（从低到高）数据库 > 缓存 > Zookeeper

从性能角度（从高到低）缓存 > Zookeeper >= 数据库

从可靠性角度（从高到低）Zookeeper > 缓存 > 数据库