JAVA分布式锁介绍

╰+攻爆jí腚メ 2024-04-17 05:35 105阅读 0赞

上一篇咱们讲到同一个进程，不同的线程之间我们可以通过synchronized、ReentrantLock、ReadWriteLock、Semaphore、CountDownLatch这些来实现锁机制。现在情况不一样了，咱们的程序高大上了，咱们可以部署多个服务端了，上了分布式系统了。在这个时候锁就要上升一个档次了，现在就叫分布式锁了。

分布式锁就是在分布式系统中(多服务端，多进程)中保证数据的最终唯一性。当在分布式模型下，数据只有一份(或有限制)，此时需要利用锁的技术控制某一时刻修改数据的进程数。现在市面上常见的有三种实现分布式锁的方案：基于数据库做分布式锁、基于redis做分布式锁、基于zookeeper做分布式锁。

# 一 基于数据库做分布式锁 #

## 1.1 基于主键的唯一性做分布式锁(悲观锁) ##

既然是通过数据库来做分布式锁，那咱们的先建一张表先，稍后我们在做解释。

> unionkeylock表（resource\_name唯一索引）

CREATE TABLE `unionkeylock` (
    	`id` int(11) unsigned NOT NULL AUTO_INCREMENT,
    	`resource_name` varchar(128) NOT NULL DEFAULT '' COMMENT '锁定的资源',
    	`node_info` varchar(128) DEFAULT NULL COMMENT '机器信息',
    	`count` int(11) NOT NULL DEFAULT '0' COMMENT '锁的次数,统计可重入锁',
    	`des` varchar(128) DEFAULT null comment '额外的描述信息',
    	`update_time` timestamp NULL DEFAULT NULL COMMENT '更新时间',
    	`create_time` timestamp null default null comment '创建时间',
    	primary key (`id`),
        unique key (`resource_name`)
    ) engine=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='数据库分布式锁表-主键唯一';;

unionkeylock表，我们对resource\_name做了唯一性约束，如果有多个请求同时提交到数据库的话，数据库会保证只有一个操作可以成功，在这种情况下我们认为操作成功的那个线程获得了该方法的锁。

我们干脆一点，直接上代码。

public class DbDistributedUnionKeyLock extends AbstractDbDistributedLock {
        /**
         * 因为数据库里面限制了最大长度了
         */
        private static final int NODE_INFO_MAX_LENGTH = 120;
    
        /**
         * 计算机唯一标识 为可重入锁做准备(一定要确保每个机器的值不同)
         */
        private static final String COMPUTER_UUID = ComputerIdentifierUtil.getComputerIdentifier();
        /**
         * 线程变量
         */
        private ThreadLocal<String> threadFlag = new ThreadLocal<>();
        /**
         * 操作数据库的dao
         */
        private IUnionKeyLockDao unionKeyLockDao;
    
        public DbDistributedUnionKeyLock(IUnionKeyLockDao unionKeyLockDao) {
            this.unionKeyLockDao = unionKeyLockDao;
        }
    
        /**
         * 加锁
         */
        @Override
        public boolean lock(String key, int retryTimes, long sleepMillis) {
            boolean lockSuccess = false;
            // 机器码+线程uuid -- 唯一标识（保证同一台电脑的同一个线程是一样的）
            if (threadFlag.get() == null || threadFlag.get().isEmpty()) {
                String nodeTemp = COMPUTER_UUID + "#" + String.format("%08x", UUID.randomUUID().hashCode()) + "#" + Thread.currentThread().getId();
                if (nodeTemp.length() > NODE_INFO_MAX_LENGTH) {
                    nodeTemp = nodeTemp.substring(0, NODE_INFO_MAX_LENGTH);
                }
                threadFlag.set(nodeTemp);
            }
            int retry = 0;
            while (!lockSuccess && retry < retryTimes) {
                try {
                    UnionKeyLock lockInfo = unionKeyLockDao.getLockInfoByResourceName(key);
                    if (lockInfo == null) {
                        // 当前资源没有被加锁
                        lockSuccess = unionKeyLockDao.insertLockInfo(key, threadFlag.get());
                    } else {
                        // 当前资源已经被加锁,这个时候需要考虑是否可重入
                        // 可重入锁
                        if (lockInfo.getNodeInfo() != null && lockInfo.getNodeInfo().equals(threadFlag.get())) {
                            // 重入
                            lockSuccess = unionKeyLockDao.reentrantLock(lockInfo.getResourceName(), threadFlag.get(), lockInfo.getCount());
                        } else {
                            // 如果出现这种情况代表数据已经有问题了
                            lockSuccess = false;
                        }
                    }
                    if (!lockSuccess) {
                        // 等待200毫秒
                        Uninterruptibles.sleepUninterruptibly(sleepMillis, TimeUnit.MILLISECONDS);
                    }
                } catch (Exception e) {
                    // 等待指定时间
                    Uninterruptibles.sleepUninterruptibly(sleepMillis, TimeUnit.MILLISECONDS);
                } finally {
                    retry++;
                }
            }
            return lockSuccess;
        }
    
    
        /**
         * 释放
         */
        @Override
        public void unlock(String key) {
            if (threadFlag.get() == null || threadFlag.get().isEmpty()) {
                return;
            }
            boolean unlockSuccess = false;
            while (!unlockSuccess) {
                try {
                    UnionKeyLock lockInfo = unionKeyLockDao.getLockInfoByResourceName(key);
                    if (lockInfo == null) {
                        return;
                    }
                    if (lockInfo.getNodeInfo() == null) {
                        unionKeyLockDao.deleteLockInfo(lockInfo.getResourceName(), lockInfo.getNodeInfo());
                        return;
                    }
                    if (!lockInfo.getNodeInfo().equals(threadFlag.get())) {
                        return;
                    }
                    // 可重入锁
                    if (lockInfo.getCount() == 1) {
                        unionKeyLockDao.deleteLockInfo(lockInfo.getResourceName(), threadFlag.get());
                        unlockSuccess = true;
                    } else {
                        if (lockInfo.getNodeInfo() != null && lockInfo.getNodeInfo().equals(threadFlag.get())) {
                            // 重入
                            unlockSuccess = unionKeyLockDao.reentrantUnLock(lockInfo.getResourceName(), threadFlag.get(), lockInfo.getCount());
                        } else {
                            // 如果出现这种情况，代表这时候数据已经有问题了
                            unionKeyLockDao.deleteLockInfo(lockInfo.getResourceName(), lockInfo.getNodeInfo());
                            unlockSuccess = true;
                        }
                    }
                    if (!unlockSuccess) {
                        // 等待200毫秒
                        Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
                    }
                } catch (Exception e) {
    //                e.printStackTrace();
                    // 等待200毫秒
                    Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
                }
    
            }
    
        }
    
    }

> 因为悲观锁大多数情况下依靠数据库的锁机制实现，以保证操作最大程度的独占性。如果加锁的时间过长，其他用户长时间无法访问，影响了程序的并发访问性，同时这样对数据库性能开销影响也很大，特别是对长事务而言，这样的开销往往无法承受。所以与悲观锁相对的，我们有了乐观锁。

## 1.2 基于版本号字段做分布式锁(乐观锁) ##

基于表字段版本号做分布式锁(乐观锁)：很乐观，每次去拿数据的时候都认为别人不会修改，所以不会上锁，当读取数据时，将version字段的值一同读出，数据每更新一次，对此version值加一。当我们提交更新的时候，需要数据库表对应记录的当前版本信息与第一次取出来的version值进行比对，如果数据库表当前版本号与第一次取出来的version值相等，则予以更新，否则重试。乐观锁适用于读多写少的应用场景，这样可以提高吞吐量。

通过版本号做分布式锁我们不太好去抽象出lock()、unlock()这样的函数了。得把他们写到业务逻辑里面去。下面我们以一个简单的例子来说明。比如有这么个场景：我们模拟一个存钱的场景，我们可以在各个地方存钱(分布式)。我们创建一个表，字段value就表示我们的余额。

CREATE TABLE `optimisticlock` (
    	`id` int(11) unsigned NOT NULL AUTO_INCREMENT,
    	`resource_name` varchar(128) NOT NULL DEFAULT '' COMMENT '锁定的资源',
    	`value` int NOT NULL COMMENT '锁定的资源对应的值',
    	`version` int NOT NULL COMMENT '版本信息',
    	PRIMARY KEY (`id`),
    	UNIQUE KEY `uiq_idx_resource` (`resource_name`) 
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='数据库分布式锁表-乐观锁';

直接上代码。(可以在[https://github.com/tuacy/java-study][https_github.com_tuacy_java-study]里面找到更加完整的代码，注意哦代码在distributedlock模块下面哦)

public class SqlOptimisticLock {
    
        /**
         * 因为数据库里面限制了最大长度了
         */
        private static final int LOCK_LOOP_TIME_WAIT = 200;
        /**
         * lock对应的资源名字
         */
        private final String resourceName;
        /**
         * 操作数据库的dao
         */
        private IOptimisticLockDao optimisticLockDao;
    
        /**
         * 用于注入无法自动装配的Bean,处理@Autowired无效的问题
         */
        private void inject() {
            if (null == this.optimisticLockDao) {
                this.optimisticLockDao = ResourceApplicationContext.getApplicationContext().getBean(OptimisticLockDaoImpl.class);
            }
        }
    
        /**
         * 构造函数
         *
         * @param resourceName lock对应资源名字
         */
        public SqlOptimisticLock(String resourceName) {
            this.resourceName = resourceName;
            inject();
        }
    
        /**
         * 我们简单的模拟一个存钱的操作
         *
         * @param money 存入金额
         */
        public void depositMoney(int money) {
            boolean success = false;
            while (!success) {
                try {
                    // 第一步：版本号信息很重要，从表里面取出数据(包括版本号)
                    OptimisticLock dbItemInfo = optimisticLockDao.selectLockResourceInfo(resourceName);
                    // 第二步：todo:做相应的逻辑处理
                    Uninterruptibles.sleepUninterruptibly(2, TimeUnit.SECONDS);
                    // 第三步：把数据存到数据库里面去
                    if (dbItemInfo == null) {
                        success = optimisticLockDao.insertLockResourceValue(resourceName, money);
                    } else {
                        // 更新的时候会去做下版本的判断，相同才更新，不相同不更新
                        success = optimisticLockDao.updateLockResourceValue(resourceName, dbItemInfo.getVersion(), dbItemInfo.getValue() + money);
                    }
                    if (!success) {
                        Uninterruptibles.sleepUninterruptibly(LOCK_LOOP_TIME_WAIT, TimeUnit.MILLISECONDS);
                    }
                } catch (Exception e) {
    //                e.printStackTrace();
                    success = false;
                    Uninterruptibles.sleepUninterruptibly(LOCK_LOOP_TIME_WAIT, TimeUnit.MILLISECONDS);
    
                }
    
            }
        }
    
    }

## 1.3 基于数据库排他锁做分布式锁(悲观锁) ##

借助数据中自带的排他锁来实现分布式的锁。在查询语句后面增加for update，数据库会在查询过程中给数据库表增加排他锁。当某条记录被加上排他锁之后，其他线程无法再在该行记录上增加排他锁。特别不建议用这种方式来实现，对数据库影响太大了。

正对这种情况，我们也写一个简单的代码，还是先创建表，建表语句如下。

CREATE TABLE `exclusivelock` (
    	`id` int(11) unsigned NOT NULL AUTO_INCREMENT,
    	`resource_name` varchar(128) NOT NULL DEFAULT '' COMMENT '锁定的资源',
    	PRIMARY KEY (`id`),
    	UNIQUE KEY `uiq_idx_resource` (`resource_name`) 
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='数据库分布式锁表-排他锁';

加锁，释放锁代码实现。(可以在[https://github.com/tuacy/java-study][https_github.com_tuacy_java-study]里面找到更加完整的代码，注意哦代码在distributedlock模块下面哦)

public class SqlExclusiveLock implements IDistributedLock {
    
        /**
         * 因为数据库里面限制了最大长度了
         */
        private static final int LOCK_LOOP_TIME_WAIT = 200;
    
        private JdbcTemplate jdbcTemplate;
        /**
         * lock对应的资源名字
         */
        private final String resourceName;
    
        private Connection sqlConnection;
    
        /**
         * 用于注入无法自动装配的Bean,处理@Autowired无效的问题
         */
        private void inject() {
            if (null == this.jdbcTemplate) {
                this.jdbcTemplate = ResourceApplicationContext.getApplicationContext().getBean(JdbcTemplate.class);
            }
        }
    
        /**
         * 构造函数
         *
         * @param resourceName lock对应资源名字
         */
        public SqlExclusiveLock(String resourceName) {
            this.resourceName = resourceName;
            inject();
        }
    
        @Override
        public void lock() {
            if (jdbcTemplate.getDataSource() == null) {
                throw new IllegalArgumentException("数据库配置失败!");
            }
            boolean success = false;
            while (!success) {
                PreparedStatement preparedStatement = null;
                try {
                    sqlConnection = jdbcTemplate.getDataSource().getConnection();
                    sqlConnection.setAutoCommit(false);//设置手动提交
                    String prepareSql = "select * from exclusivelock where resource_name = ? for update";
                    preparedStatement = sqlConnection.prepareStatement(prepareSql);
                    preparedStatement.setString(1, this.resourceName);
                    success = preparedStatement.executeQuery() != null;
                    if (!success) {
                        Uninterruptibles.sleepUninterruptibly(LOCK_LOOP_TIME_WAIT, TimeUnit.MILLISECONDS);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    success = false;
                    Uninterruptibles.sleepUninterruptibly(LOCK_LOOP_TIME_WAIT, TimeUnit.MILLISECONDS);
                } finally {
                    if (preparedStatement != null) {
                        try {
                            preparedStatement.close();
                        } catch (SQLException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
        }
    
        @Override
        public void unlock() {
            if (sqlConnection != null) {
                try {
                    sqlConnection.commit();
                } catch (Exception e) {
                    // ignore
                    e.printStackTrace();
                }
            }
        }
    
        @Override
        public boolean tryLock() {
            return false;
        }
    
        @Override
        public boolean tryLock(long time, TimeUnit unit) {
            return false;
        }
    }

# 二 基于Redis做分布式锁 #

基于Redis做分布式锁。Redis里面的记录可以设置为当记录不存在的时候才可以插入进入。记录存在则插入不进去。Redis分布式锁就是用这个做的。当获取锁的时候则在Redis里面插入一条这样的记录。释放锁的时候就删除这条记录。如果锁没有释放(记录没有被删除)其他的让你再去获取锁的时候(插入记录)是不会成功的。而且为了防止僵尸我们可以给这条记录设置过期时间。

讲什么都没带来来的实在点。我们就直接贴代码了(可以在[https://github.com/tuacy/java-study][https_github.com_tuacy_java-study]里面找到更加完整的代码，注意哦代码在distributedlock模块下面哦)

public class RedisDistributedLockImpl extends AbstractDistributedLock {
    
        private final Logger logger = LoggerFactory.getLogger(RedisDistributedLockImpl.class);
    
        private RedisTemplate<Object, Object> redisTemplate;
    
        private ThreadLocal<String> lockFlag = new ThreadLocal<>();
    
        /**
         * Lua
         */
        private static final String UNLOCK_LUA;
    
        static {
            StringBuilder sb = new StringBuilder();
            sb.append("if redis.call(\"get\",KEYS[1]) == ARGV[1] ");
            sb.append("then ");
            sb.append("    return redis.call(\"del\",KEYS[1]) ");
            sb.append("else ");
            sb.append("    return 0 ");
            sb.append("end ");
            UNLOCK_LUA = sb.toString();
        }
    
        /**
         * 构造函数
         */
        public RedisDistributedLockImpl(RedisTemplate<Object, Object> redisTemplate) {
            super();
            this.redisTemplate = redisTemplate;
        }
    
        /**
         * 加锁
         */
        @Override
        public boolean lock(String key, long expire, int retryTimes, long sleepMillis) {
            boolean result = setRedis(key, expire);
            // 如果获取锁失败，按照传入的重试次数进行重试
            while ((!result) && retryTimes-- > 0) {
                try {
                    logger.debug("lock failed, retrying..." + retryTimes);
                    Thread.sleep(sleepMillis);
                } catch (InterruptedException e) {
                    return false;
                }
                result = setRedis(key, expire);
            }
            return result;
        }
    
        private boolean setRedis(String key, long expire) {
            try {
                String result = redisTemplate.execute((RedisCallback<String>) connection -> {
                    JedisCommands commands = (JedisCommands) connection.getNativeConnection();
                    // value生成
                    String uuid = UUID.randomUUID().toString();
                    lockFlag.set(uuid);
                    /**
                     * 存储数据到缓存中，并指定过期时间和当Key存在时是否覆盖。
                     *
                     * @param key 键
                     * @param key 键值
                     * @param nxxx
                     *            nxxx的值只能取NX或者XX，如果是NX的时候，则只有当key不存在是才进行set，如果是XX，则只有当key已经存在时才进行set
                     *
                     * @param expx expx的值只能取EX或者PX，代表数据过期时间的单位，EX代表秒，PX代表毫秒。
                     * @param time 过期时间，单位是expx所代表的单位。
                     * @return 成功返回“ok”，失败则返回 null。
                     */
                    return commands.set(key, uuid, "NX", "PX", expire);
                });
                return !StringUtils.isEmpty(result);
            } catch (Exception e) {
                logger.error("set redis occured an exception", e);
            }
            return false;
        }
    
        /**
         * 释放锁
         */
        @Override
        public boolean unlock(String key) {
            // 释放锁的时候，有可能因为持锁之后方法执行时间大于锁的有效期，此时有可能已经被另外一个线程持有锁，所以不能直接删除
            try {
                List<String> keys = new ArrayList<>();
                keys.add(key);
                List<String> args = new ArrayList<>();
                args.add(lockFlag.get());
    
                // 使用lua脚本删除redis中匹配value的key，可以避免由于方法执行时间过长而redis锁自动过期失效的时候误删其他线程的锁
                // spring自带的执行脚本方法中，集群模式直接抛出不支持执行脚本的异常，所以只能拿到原redis的connection来执行脚本
    
                Long result = redisTemplate.execute((RedisCallback<Long>) connection -> {
                    Object nativeConnection = connection.getNativeConnection();
                    // 集群模式和单机模式虽然执行脚本的方法一样，但是没有共同的接口，所以只能分开执行
                    // 集群模式
                    if (nativeConnection instanceof JedisCluster) {
                        // Redis Eval 命令使用 Lua 解释器执行脚本
                        return (Long) ((JedisCluster) nativeConnection).eval(UNLOCK_LUA, keys, args);
                    }
                    // 单机模式
                    else if (nativeConnection instanceof Jedis) {
                        // Redis Eval 命令使用 Lua 解释器执行脚本
                        return (Long) ((Jedis) nativeConnection).eval(UNLOCK_LUA, keys, args);
                    }
                    return 0L;
                });
    
                return result != null && result > 0;
            } catch (Exception e) {
                logger.error("release lock occured an exception", e);
            }
            return false;
        }
    }

# 三 基于ZooKeeper做分布式锁 #

Zookeeper实现分布式锁的流程，假设锁空间的根节点为/zklock：

*  客户端连接zookeeper，并在/zklock下创建临时的且有序的子节点。  
    第一个客户端对应的子节点为：/zklock/test\_lock\_0000000000，第二个为：/zklock/test\_lock\_0000000001。以此类推。
 *  客户端获取/zklock下的子节点列表，判断自己创建的子节点是否为当前子节点列表中序号最小的子节点，如果是则认为获得锁，否则监听/zklock的子节点变更消息，获得子节点变更通知后重复此步骤直至获得锁;
 *  执行业务代码。
 *  完成业务流程后，删除对应的子节点并释放锁。

这里我不准备用zookeeper的原生api来实现这一套流程来实现分布式锁。因为我找到更简单的方式，这里我准备用Curator(Curator是Netflix公司开源的一套zookeeper客户端框架)来实现分布式锁。因为Curator已经帮我们实现了分布式锁InterProcessMutex的实现。关于Curator的使用大家有兴趣可以参考下[ZooKeeper客户端Curator使用][ZooKeeper_Curator]

直接代码实现，(可以在[https://github.com/tuacy/java-study][https_github.com_tuacy_java-study]里面找到更加完整的代码，注意哦代码在distributedlock模块下面哦)

public class ZookeeperDistributedLockImpl implements IZookeeperDistributedLock {
    
        /**
         * curator给封住好的一个分布式锁对象和ReentrantLock类似
         */
        private CuratorFramework curatorFramework;
    
        public ZookeeperDistributedLockImpl(ZkClient zkClient) {
            curatorFramework = zkClient.getClient();
        }
    
        @Override
        public boolean lock(String key) {
            try {
                InterProcessMutex  interProcessMutex = new InterProcessMutex(curatorFramework, key);
                interProcessMutex.acquire();
                return true;
            } catch (Exception e) {
                e.printStackTrace();
                return false;
            }
        }
    
        @Override
        public boolean lock(String key, long time, TimeUnit unit) {
            try {
                InterProcessMutex  interProcessMutex = new InterProcessMutex(curatorFramework, key);
                interProcessMutex.acquire(time, unit);
                return true;
            } catch (Exception e) {
                e.printStackTrace();
                return false;
            }
        }
    
        @Override
        public void unlock(String key) {
            try {
                InterProcessMutex  interProcessMutex = new InterProcessMutex(curatorFramework, key);
                interProcessMutex.release();
            } catch (Exception e) {
                e.printStackTrace();
                // ignore
            }
        }

--------------------

分布式锁相关的内容就介绍到这里，每种分布式锁的时候我们都写了对应的实现代码。大家可以在[https://github.com/tuacy/java-study][https_github.com_tuacy_java-study]里面找到更加完整的代码，在distributedlock模块下面。而且每种锁的实现都用想用的AOP。

[https_github.com_tuacy_java-study]: https://github.com/tuacy/java-study
[ZooKeeper_Curator]: https://blog.csdn.net/wuyuxing24/article/details/99710608