在一次工作中，出现了一个小意外，一个存在redis写与读的方法中得出的结果总是和预想的不一样，后面仔细思考了以后，发现是因为集群中各个节点都使用共享的缓存、队列这些，有些场景中各个节点之间可能会发生资源竞争，可能会发各个节点之间的“线程不安全问题”，单机中，可以使用锁来解决，在分布式环境下，就要用到分布式锁了，当时网上查了查，发现了好多大神给出的思路和代码，自己试着实现了一下，在这把思路分享给大家。
Redis可以做分布式锁的条件
因为redis是单线程的，并行访问的指令在他内部会是串行执行的，所以redis可以作为实现分布式锁的技术。
所用的命令
在前几篇文章中，已经介绍了Redis的基本类型与其相应的操作，有不太了解的朋友可以去《Redis常用类型知多少？》去看看。在实现分布式锁中，主要用到了redis的字符串类型数据结构，以及以下的操作:
set key value [ex 秒数] [px 毫秒数] [nx/xx] : 保存key-value
ex、px指的是key的有效期。
nx：表示当key不存在时，创建key并保存value
xx：表示当key存在时，保存value
localhost:0>set hello world
OK
localhost:0>set hello world ex 100
OK
localhost:0>set hello world px 1000
OK
这里主要看看 nx与xx的效果
localhost:0>set hello worldnx nx
NULL
localhost:0>set hello worldxx xx
OK
因为hello这个key已经存在，所以带上参数nx后，返回了NULL，没有操作成功。
带上xx这个参数后，现在hello这个key的value已经被替换成了worldxx
复制代码实现中用到了set指令的nx参数，作用是保存一个key-value，如果key不存在时，创建key并保存value，如果key存在是，不做操作返回NULL，在java代码中，这个命令被封装成了Jedis的一个方法，我自己在代码中又做了一个封装：
/**
* 如果key不存在则建立返回1
* 如果key存则不作操作，返回0
* @param key
* @param value
* @return
*/
public Long setNx(final String key,final String value){
return this.execute(new Function<Long, Jedis>() {
@Override
public Long callback(Jedis jedis) {
//set命令带nx参数在Jedis中被封装成了setnx方法
return jedis.setnx(key, value);
}
});
}
复制代码主要用来获取锁，setnx返回1，证明获取到了锁，返回0证明锁已经存在。
expire key 整数值：设置key的生命周期以秒为单位
主要用来设置锁的过期时间，贴出java中封装的方法expire：
/**
* 设置key的过期时间
* @param key
* @param exp
* @return
*/
public Long expire(final String key,final int exp){
return this.execute(new Function<Long, Jedis>() {
@Override
public Long callback(Jedis jedis) {
return jedis.expire(key, exp);
}
});
}
复制代码get key : 取出key对应的value
取出存进去的锁值，贴出java中封装的方法get：
/**
* 获取String类型的值
*
* @param key
* @return
*/
public String get(final String key) {
return this.execute(new Function<String, Jedis>() {
@Override
public String callback(Jedis jedis) {
return jedis.get(key);
}
});
}
复制代码getset key newvalue：返回key的旧value，把新值newvalue存进去
localhost:0>getset hello wolrd
wogldxx
localhost:0>get hello
wolrd
复制代码用于获取到锁以后把新的锁值存进去，贴出java中封装的方法getset：
/**
* 获取并返回旧值，在设置新值
* @param key
* @param value
* @return
*/
public String getSet(final String key, final String value){
return this.execute(new Function<String, Jedis>() {
@Override
public String callback(Jedis jedis) {
return jedis.getSet(key, value);
}
});
}
复制代码实现的思路
1.先利用当前的时间戳与锁的过期时间相加后得出锁将要过期的时间，并把这个时间作为value，锁作为key调用setnx方法，如果返回1，证明原来这个key，也就是锁不存在，获取锁成功，则调用expire方法设置锁的超时时间，返回获取锁成功。
2.如果setnx返回0，证明原来锁存在，没有获取到锁，然后阻塞等锁的获取。（我原来自己的思路就考虑到第2步就结束了，后面看了大神的实现思路，发现我的想法欠妥，下面说一下大神的第3步）。
3.如果setnx返回0，证明原来锁存在，没有获取到锁，然后调用get方法，获取第1步存进去的value，就是锁将要过期的时间，与当前的时间戳比较，如果如果当前的时间戳大于锁将要过期的时间，证明锁已经过期，调用getset方法，把新的时间存进去，返回获取锁成功。这样做主要是应对expire执行失败，或者服务器重启的情况下出现的锁无法释放的情况。
下面把按照大神思路实现的代码贴出来，实现了阻塞机制的锁，也实现了一个排它锁：
import com.eduapi.common.component.RedisComponent;
import com.eduapi.common.util.BeanUtils;
import org.apache.commons.lang3.StringUtils;
/**
* @Description: 利用redis实现分布式锁.
* @Author: ZhaoWeiNan .
* @CreatedTime: 2017/3/20 .
* @Version: 1.0 .
*/
public class RedisLock {
private RedisComponent redisComponent;
private static final int DEFAULT_ACQUIRY_RESOLUTION_MILLIS = 100;
private String lockKey;
/**
* 锁超时时间，防止线程在入锁以后，无限的执行等待
*/
private int expireMillisCond = 60 * 1000;
/**
* 锁等待时间，防止线程饥饿
*/
private int timeoutMillisCond = 10 * 1000;
private volatile boolean isLocked = false;
public RedisLock(RedisComponent redisComponent, String lockKey) {
this.redisComponent = redisComponent;
this.lockKey = lockKey;
}
public RedisLock(RedisComponent redisComponent, String lockKey, int timeoutMillisCond) {
this(redisComponent, lockKey);
this.timeoutMillisCond = timeoutMillisCond;
}
public RedisLock(RedisComponent redisComponent, String lockKey, int timeoutMillisCond, int expireMillisCond) {
this(redisComponent, lockKey, timeoutMillisCond);
this.expireMillisCond = expireMillisCond;
}
public RedisLock(RedisComponent redisComponent, int expireMillisCond, String lockKey) {
this(redisComponent, lockKey);
this.expireMillisCond = expireMillisCond;
}
public String getLockKey() {
return lockKey;
}
/**
* 获得 lock. （把大神的思路粘过来了）
* 实现思路: 主要是使用了redis 的setnx命令,缓存了锁.
* reids缓存的key是锁的key,所有的共享, value是锁的到期时间(注意:这里把过期时间放在value了,没有时间上设置其超时时间)
* 执行过程:
* 1.通过setnx尝试设置某个key的值,成功(当前没有这个锁)则返回,成功获得锁
* 2.锁已经存在则获取锁的到期时间,和当前时间比较,超时的话,则设置新的值
*
* @return true if lock is acquired, false acquire timeouted
* @throws InterruptedException in case of thread interruption
*/
public synchronized boolean lock() throws InterruptedException {
int timeout = timeoutMillisCond;
boolean flag = false;
while (timeout > 0){
//设置所得到期时间
Long expires = System.currentTimeMillis() + expireMillisCond;
String expiresStr = BeanUtils.convertObject2String(expires);
//原来redis里面没有锁，获取锁成功
if (this.redisComponent.setNx(lockKey,expiresStr) > 0){
//设置锁的过期时间
this.redisComponent.expire(lockKey,expireMillisCond);
isLocked = true;
return true;
}
flag = compareLock(expiresStr);
if (flag){
return flag;
}
timeout -= DEFAULT_ACQUIRY_RESOLUTION_MILLIS;
/*
延迟100 毫秒, 这里使用随机时间可能会好一点,可以防止饥饿进程的出现,即,当同时到达多个进程,
只会有一个进程获得锁,其他的都用同样的频率进行尝试,后面有来了一些进程,也以同样的频率申请锁,这将可能导致前面来的锁得不到满足.
使用随机的等待时间可以一定程度上保证公平性
*/
Thread.sleep(DEFAULT_ACQUIRY_RESOLUTION_MILLIS);
}
return false;
}
/**
* 排他锁。作用相当于 synchronized 同步快
* @return
* @throws InterruptedException
*/
public synchronized boolean excludeLock() {
//设置所得到期时间
long expires = System.currentTimeMillis() + expireMillisCond;
String expiresStr = BeanUtils.convertObject2String(expires);
//原来redis里面没有锁，获取锁成功
if (this.redisComponent.setNx(lockKey,expiresStr) > 0){
//设置锁的过期时间
this.redisComponent.expire(lockKey,expireMillisCond);
isLocked = true;
return true;
}
return compareLock(expiresStr);
}
/**
* 比较是否可以获取锁
* 锁超时时 获取
* @param expiresStr
* @return
*/
private boolean compareLock(String expiresStr){
//假如两个线程走到这里
//因为redis是单线程的获取到
// A线程获取 currentValueStr = 1 B线程获取 currentValueStr = 1
String currentValueStr = this.redisComponent.get(lockKey);
//锁
if (StringUtils.isNotEmpty(currentValueStr) && Long.parseLong(currentValueStr) < System.currentTimeMillis()){
//获取上一个锁到期时间，并设置现在的锁到期时间，
//只有一个线程才能获取上一个线程的设置时间，因为jedis.getSet是同步的
//只有A线程 把 2 存进去了。 取出了 1， 对比获得了锁
//B线程 吧 2存进去了。 获取 2.。对比 没有获得锁，
String oldValue = this.redisComponent.getSet(lockKey,expiresStr);
if (StringUtils.isNotEmpty(oldValue) && StringUtils.equals(oldValue,currentValueStr)){
//防止误删（覆盖，因为key是相同的）了他人的锁——这里达不到效果，这里值会被覆盖，但是因为什么相差了很少的时间，所以可以接受
//[分布式的情况下]:如过这个时候，多个线程恰好都到了这里，但是只有一个线程的设置值和当前值相同，他才有权利获取锁
isLocked = true;
return true;
}
}
return false;
}
/**
* 释放锁
*/
public synchronized void unlock(){
if (isLocked){
this.redisComponent.delete(lockKey);
isLocked = false;
}
}
}
复制代码代码中，把大神写的思路站进去了，大神讲的还是很明白的，看看调用的代码：
//创建锁对象， redisComponent 为redis组件的对象 过期时间 锁的key
RedisLock redisLock = new RedisLock(redisComponent,1000 * 60,RedisCacheKey.REDIS_LOCK_KEY + now_mm);
//获取锁
if (redisLock.excludeLock()){
try {
//拿到了锁，读取定时短信有序集合
set = this.redisComponent.zRangeByScore(RedisCacheKey.MSG_TIME_LIST,0,end);
if (set != null && set.size() > 0){
flag = true;
}
} catch (Exception e) {
LOGGER.error("获取定时短信有序集合异常，异常为{}",e.toString());
}
...
复制代码利用redis实现分布式锁的一种思路就为大家介绍到这里，欢迎大家来交流，指出文中一些说错的地方，让我加深认识。
谢谢大家！
近期评论