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分布式锁设计成注解模式减少与业务代码的耦合，当然那都是后话了。
如果还有些不明白可以移步到此处:分布式锁
以上。
