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

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

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

不加锁代码：

/**
     * 减少库存-不加锁-会存在超卖问题
     *
     * @param productId
     * @return
     */
    @Transactional
    @Override
    public 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
    @Override
    public 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
    @Override
    public 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要带上version
        Integer 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
     */
    @Override
    public 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.构建RedissonClient
            RedissonClient 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();
            }
        }
    }


    /**
     * 获取锁
     */
    @Override
    public 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() {
                @Override
                public void handleDataChange(String dataPath, Object data) throws Exception {

                }

                @Override
                public 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
     */
    @Override
    public 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
     */
    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return false;
    }

    /**
     * 解锁
     */
    @Override
    public 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
     */
    @Override
    public Condition newCondition() {
        return null;
    }

    /**
     * 如果当前线程未被中断，则获取锁，可以响应中断
     *
     * @throws InterruptedException
     */
    @Override
    public void lockInterruptibly() throws InterruptedException {

    }

}

简单总结：

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

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

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