Redis实现分布式锁
Posted Andyzty
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Redis实现分布式锁相关的知识,希望对你有一定的参考价值。
Redis实现分布式锁
在集群等多服务器中经常要使用到同步处理一下业务,这时普通的事务是满足不要业务需求,需要分布式锁。分布式锁的实现方式有多种,如redis实现分布式锁,zookeeper实现分布式锁等,这篇先实现redis分布式锁。
实现原理
1、通过setnx(lock_timeout)实现,如果设置了锁返回1,已经有值没有设置成功返回0。
2、死锁问题:通过时间来判断是否过期,如果已经过期,获取到过期时间get(lockKey),然后getset(lock_timeout)判断是否和get相同,相同则证明已经加锁成功,因为可能会导致多个线程同时执行getset(lock_timeout)方法。这是可能导致多个线程都只需getset后,对于判断加锁成功的线程,再加expire(lockKey, LOCK_TIMEOUT, TimeUnit.MILLISECONDS)过期时间,防止多个线程同时叠加时间,导致锁时效时间翻倍。
3、针对集群服务器时间不一致问题,可以从调用redis的time()获取当前时间。
代码实现
分布式锁接口
DistributionLock.java
/**
* 创建时间:2016年12月8日 下午6:51:51
*
* 分布式锁
*
* @author andy
* @version 2.2
*/
public interface DistributionLock {
//加锁成功 返回加锁时间
public Long lock(String lockKey, String threadname);
//解锁 需要更加加锁时间判断是否有权限
public void unlock(String lockKey, long lockvalue, String threadname);
}
分布式锁Redis实现
RedisDistributionLock.java
import java.io.Serializable;
import java.util.concurrent.TimeUnit;
import org.apache.log4j.Logger;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
import org.andy.utils.SpringContextHolder;
/**
* 创建时间:2016年12月8日 下午5:44:16
*
* redis分布式锁
*
* @author andy
* @version 2.2
*/
public class RedisDistributionLock implements DistributionLock{
private static final long LOCK_TIMEOUT = 60 * 1000; //加锁超时时间 单位毫秒 意味着加锁期间内执行完操作 如果未完成会有并发现象
private static final Logger LOG = Logger.getLogger(RedisDistributionLock.class); //redis锁日志
@SuppressWarnings("unchecked")
private static RedisTemplate<Serializable, Serializable> redisTemplate = (RedisTemplate<Serializable, Serializable>) SpringContextHolder
.getBean("redisTemplate");
/**
* 取到锁加锁 取不到锁一直等待直到获得锁
*/
@Override
public Long lock(String lockKey, String threadname) {
LOG.info(threadname + "开始执行加锁");
while (true) { //循环获取锁
Long lock_timeout = System.currentTimeMillis() + LOCK_TIMEOUT + 1; //锁时间
if (redisTemplate.execute(new RedisCallback<Boolean>() {
@Override
public Boolean doInRedis(RedisConnection connection) throws DataAccessException {
JdkSerializationRedisSerializer jdkSerializer = new JdkSerializationRedisSerializer();
byte[] value = jdkSerializer.serialize(lock_timeout);
return connection.setNX(lockKey.getBytes(), value);
}
})) { //如果加锁成功
LOG.info(threadname + "加锁成功++++++++111111111");
redisTemplate.expire(lockKey, LOCK_TIMEOUT, TimeUnit.MILLISECONDS); //设置超时时间,释放内存
return lock_timeout;
}else {
Long currt_lock_timeout_Str = (Long) redisTemplate.opsForValue().get(lockKey); // redis里的时间
if (currt_lock_timeout_Str != null && currt_lock_timeout_Str < System.currentTimeMillis()) { //锁已经失效
// 判断是否为空,不为空的情况下,说明已经失效,如果被其他线程设置了值,则第二个条件判断是无法执行
Long old_lock_timeout_Str = (Long) redisTemplate.opsForValue().getAndSet(lockKey, lock_timeout);
// 获取上一个锁到期时间,并设置现在的锁到期时间
if (old_lock_timeout_Str != null && old_lock_timeout_Str.equals(currt_lock_timeout_Str)) {
// 如过这个时候,多个线程恰好都到了这里,但是只有一个线程的设置值和当前值相同,他才有权利获取锁
LOG.info(threadname + "加锁成功+++++++2222222222");
redisTemplate.expire(lockKey, LOCK_TIMEOUT, TimeUnit.MILLISECONDS); //设置超时时间,释放内存
return lock_timeout;//返回加锁时间
}
}
}
try {
LOG.info(threadname + "等待加锁,睡眠100毫秒");
TimeUnit.MILLISECONDS.sleep(100);//睡眠100毫秒
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
@Override
public void unlock(String lockKey, long lockvalue, String threadname) {
LOG.info(threadname + "执行解锁==========");//正常直接删除 如果异常关闭判断加锁会判断过期时间
Long currt_lock_timeout_Str = (Long) redisTemplate.opsForValue().get(lockKey); // redis里的时间
if (currt_lock_timeout_Str != null && currt_lock_timeout_Str == lockvalue) {//如果是加锁者 则删除锁 如果不是则等待自动过期 重新竞争加锁
redisTemplate.delete(lockKey); //删除键
LOG.info(threadname + "解锁成功-----------------");
}
}
}
注:上面接口中参数threadname只是为了测试多线程数据打印,生成环境可以去掉接口DistributionLock和实现类RedisDistributionLock中的String threadname参数。
多服务器采用获取redis时间,代替上面使用到的所有当前时间System.currentTimeMillis()。
public long currtTimeFromRedis(){ //获取redis当前时间
return redisTemplate.execute(new RedisCallback<Long>() {
@Override
public Long doInRedis(RedisConnection connection) throws DataAccessException {
return connection.time();
}
});
}
模拟测试
测试代码
模拟20个线程同时执行业务,获取资源。
@Test
public void testRedisDistributionLock(){
for (int i = 0; i < 20; i++) {
new Thread(new Runnable() {
@Override
public void run() {
task(Thread.currentThread().getName());
}
}).start();
}
try {
TimeUnit.MINUTES.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
private void task(String name){
DistributionLock lock = new RedisDistributionLock();
Long locktime; //加锁时间
if ((locktime = lock.lock(RedisKeyUtil.DISTRIBUTED_LOCK_NO + 1, name)) != null) {
//开始执行任务
System.out.println(name + "任务执行中");
//任务执行完毕 关闭锁
lock.unlock(RedisKeyUtil.DISTRIBUTED_LOCK_NO + 1, locktime, name);
}
}
测试结果
加锁后,保证只有一个线程获取当前资源,释放锁后释放资源。
有时间写一下zookeeper实现分布式锁
以上是关于Redis实现分布式锁的主要内容,如果未能解决你的问题,请参考以下文章