上一篇咱们讲到同一个进程，不同的线程之间我们可以通过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里面找到更加完整的代码，注意哦代码在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里面找到更加完整的代码，注意哦代码在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里面找到更加完整的代码，注意哦代码在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使用
直接代码实现，(可以在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里面找到更加完整的代码，在distributedlock模块下面。而且每种锁的实现都用想用的AOP。
网友评论