1.问题引出
先上代码 (单机环境)
private synchronized void addRecord() {
try {
//获取redis记录
DeliveryQueryRecord record = backRedisService.getDeliveryQueryRecord();
if (record == null) {
record = new DeliveryQueryRecord();
}
// xxx数据操作
//将数据再次保存到redis中
backRedisService.setDeliveryQueryRecord(record);
} catch (Exception e) {
e.printStackTrace();
}
}
由于使用了synchronized
关键字可以保证每个线程都会同步阻塞的执行,也就是说一个线程获取redis记录
并且各种操作写入redis
之后其他线程才能读和写(针对addRecord
方法的,不是真正意义上的读写)进而保证数据正确性。
但是这么写单机环境下ok,可是线上是分布式集群环境,当不止一台机器同时多线程执行addRecord
方法就可能存在数据同步问题。例如:A机器上有一个线程1去获取redis记录
,这时同时机器B也有一个线程去获取redis记录
,虽然我们知道 redis是单线程同步阻塞 的,但是极端情况下,在线程1还没写入redis
之前,线程2可能会去获取数据,而此时的数据和线程1获取的是一模一样的,而不是获取线程1处理之后的数据,也就是说两个线程不是同步阻塞获取那么数据就是有问题的。
2.解决方法
(多机环境)
问题:
解决思路:
工具筹备
工欲善其事,必先利其器。有了牛X的工具代码,业务代码也就迎刃而解。但是在展示利器之前先展示下错误的“坑”。
加锁坑:( 错误示例)
既然已经理清了redis分布式锁实现的思路,那就上代码
public boolean tryGetLock(String lockKey, String value, int expireTime) {
Jedis redis = null;
try {
redis = redisManager.redisPool.getResource();
boolean flag = redis.exists(key);//判断key是否存在
if (!flag) {//不存在则设置key
String result = redis.set(lockKey,expireTime, value);
if (LOCK_SUCCESS.equals(result)) {
return true;
}
return false;
}
} finally {
if (null != redis) {
redis.close();
}
}
}
利器代码
private static final String LOCK_SUCCESS = "OK";
private static final String SET_IF_NOT_EXIST = "NX";
private static final String SET_WITH_EXPIRE_TIME = "EX"; //px:毫秒 ex秒
private static final Long RELEASE_SUCCESS = 1L;
@Autowired
RedisManager redisManager;
/**
* 尝试获取分布式锁
*
* @param lockKey 锁
* @param requestId 请求标识
* @param expireTime 超期时间
* @return 是否获取成功
*/
public boolean tryGetLock(String lockKey, String requestId, int expireTime) {
Jedis redis = null;
try {
redis = redisManager.redisPool.getResource();
String result = redis.set(lockKey, requestId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);
if (LOCK_SUCCESS.equals(result)) {
return true;
}
return false;
} finally {
if (null != redis) {
redis.close();
}
}
}
在redisxxx.xxx版本之后,redis支持set多参数
方法:set(KEY,VALUE,SET_IF_NOT_EXIST,SET_WITH_EXPIRE_TIME,EXPIRE_TIME)
KEY,VALUE
不必多说
SET_IF_NOT_EXIST
处理模式:NX
表示如果KEY
不存在则插入成功,否则不插入数据此处正好是把 两步操作合并,1查询key是否存在 2修改数据,这样由于redis的单线程阻塞,保证了两步操作的原子性
,即阻塞执行
SET_WITH_EXPIRE_TIME
过期时间单位: px:毫秒 ex秒
EXPIRE_TIME
过期时间:保证在宕机等其他原因不能正常删除KEY(释放锁
)时,利用过期删除,从而避免KEY没有删除而造成其他线程永久等待的死锁。
此方法完美解决了上锁坑的问题。
放锁坑:(错误示例)
public void unLock(String lockKey) {
Jedis redis = null;
try {
redis = redisManager.redisPool.getResource();
redis.del(lockKey);
} finally {
if (null != redis) {
redis.close();
}
}
}
利器代码
/**
* 释放分布式锁
* @param lockKey 锁
* @param requestId 请求标识
* @return 是否释放成功
*/
public boolean unLock(String lockKey, String requestId) {
Jedis redis = null;
try {
redis = redisManager.redisPool.getResource();
//lua脚本
String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
//执行lua脚本
Object result = redis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(requestId));
if (RELEASE_SUCCESS.equals(result)) {
return true;
}
return false;
} finally {
if (null != redis) {
redis.close();
}
}
}
其他问题:
工具
/**
* @Description: 获取锁剩余时间
* @author: wz
* @date: 2019/8/2 14:17
*/
public Long getLockExpire(String key) {
Jedis redis = null;
try {
redis = redisManager.redisPool.getResource();
return redis.ttl(key);
} finally {
if (null != redis) {
redis.close();
}
}
}
/**
* @Description: 设置key 新的过期时间
* @author: wz
* @date: 2019/8/2 11:23
*/
public boolean setLockExpire(String key, int newExpire) {
Jedis redis = null;
try {
redis = redisManager.redisPool.getResource();
return redis.expire(key, newExpire) > 0;
} finally {
if (null != redis) {
redis.close();
}
}
}
守护线程代码
//守护线程
public class DeliveryDaeThread implements Runnable {
private boolean openDaeThread = true;
public void open() {
openDaeThread = true;
}
//关闭守护线程
public void close() {
openDaeThread = false;
}
@Override
public void run() {
while (openDaeThread) {
try {
System.out.println("wait...");
//在过期时间之前 2s 续命
if (getLockExpire() <= 2) {
System.out.println("=============>续命");
if (openDaeThread) {
setLockExpire();
}
}
Thread.sleep(1000);
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
核心加锁代码
/**
* @Description: 对修改redis记录前进行加锁, 如果不能立即获取到锁,则循环等待
* @author: wz
* @date: 2019/8/2 14:58
*/
public DeliveryDaeThread lock(String lockId) throws InterruptedException {
//尝试获取分布式锁的超时时间 1min
int timeout = 100;
boolean locked = this.lockDeliveryRecord(lockId);
//没取到锁 继续尝试取锁
while (!locked) {
Thread.sleep(600);
Thread thread = Thread.currentThread();
logger.info(thread.getId() + "--" + thread.getName() + " 尝试获取分布式锁");
locked = this.lockDeliveryRecord(lockId);
timeout--;
//超时
if (timeout <= 0) {
break;
}
}
//如果获取到锁,开启守护线程
if (locked) {
DeliveryDaeThread daeThread = new DeliveryDaeThread();
Thread thread = new Thread(daeThread);
thread.setDaemon(true);
thread.start();
return daeThread;
}
//没有获取到锁
return null;
}
还有问题?
我们注意到在JDK线程锁
里有会有方法设置线程等待超时时间
,如果我们要求redis分布式锁
的性能就不得不考虑:拥有锁的线程执行完业务代码释放锁
之前,任何线程不能获取锁
。这个条件造成的阻塞。想像一个极端条件:如果一个线程超久不释放锁
那么整个分布式系统等待获取该锁的线程都将等待超久。所以这里我们还需要设置一个等待超时时间
,如果线程超时不能获取到锁,难么就会返回null
,同时继续执行下面代码。当然超时时间需要根据具体业务代码来设定一个比较中肯合适的值。
结语
终于到结语了啊!笔者此刻的感受就是肩膀超级酸有木有啊!!!从发现问题到解决问题(1d)再到码字写简书(一动不动2h)
遇到问题要充满惊喜,多思考,多动手,最终解决掉,嗯。
将来有打算利用springAOP将redis分布式锁
设计成注解模式
减少与业务代码的耦合,当然那都是后话了。
如果还有些不明白可以移步到此处:分布式锁
以上。