分布式系统定时任务,保证只有一个服务执行了改任务--采用redis分布式锁来实现(文章摘自:https://www.cnblogs.com/0201zcr/p/5942748.html)

时间:2022-03-08 13:36:01

文章摘自:https://www.cnblogs.com/0201zcr/p/5942748.html

package com.abtc.server.mine.common.utils;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
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.StringRedisSerializer; import java.util.UUID; public class RedisLock
{
private static Logger logger = LoggerFactory.getLogger(RedisLock.class); private RedisTemplate redisTemplate; private static final int DEFAULT_ACQUIRY_RESOLUTION_MILLIS = 10; // 记录当前获取锁的标识
private String uuid = UUID.randomUUID().toString(); /**
* Lock key path.
*/
private String lockKey; /**
* 锁超时时间,防止线程在入锁以后,无限的执行等待
*/
private int expireMsecs = 60 * 1000; /**
* 锁等待时间,防止线程饥饿
*/
private int timeoutMsecs = 10 * 1000; private volatile boolean locked = false; /**
* Detailed constructor with default acquire timeout 10000 msecs and lock expiration of 60000 msecs.
*
* @param lockKey lock key (ex. account:1, ...)
*/
public RedisLock(RedisTemplate redisTemplate, String lockKey)
{
this.redisTemplate = redisTemplate;
this.lockKey = lockKey + "_lock";
this.uuid = uuid;
} /**
* Detailed constructor with default lock expiration of 60000 msecs.
*/
public RedisLock(RedisTemplate redisTemplate, String lockKey, int timeoutMsecs)
{
this(redisTemplate, lockKey);
this.timeoutMsecs = timeoutMsecs;
} /**
* Detailed constructor.
*/
public RedisLock(RedisTemplate redisTemplate, String lockKey, int timeoutMsecs, int expireMsecs)
{
this(redisTemplate, lockKey, timeoutMsecs);
this.expireMsecs = expireMsecs;
} /**
* @return lock key
*/
public String getLockKey()
{
return lockKey;
} private String get(final String key)
{
Object obj = null;
try
{
obj = redisTemplate.execute(new RedisCallback<Object>()
{
@Override
public Object doInRedis(RedisConnection connection) throws DataAccessException
{
StringRedisSerializer serializer = new StringRedisSerializer();
byte[] data = connection.get(serializer.serialize(key));
connection.close();
if (data == null)
{
return null;
}
return serializer.deserialize(data);
}
});
}
catch (Exception e)
{
logger.error("get redis error, key : {}", key);
}
return obj != null ? obj.toString() : null;
} private boolean setNX(final String key, final String value)
{
Object obj = null;
try
{
obj = redisTemplate.execute(new RedisCallback<Object>()
{
@Override
public Object doInRedis(RedisConnection connection) throws DataAccessException
{
StringRedisSerializer serializer = new StringRedisSerializer();
Boolean success = connection.setNX(serializer.serialize(key), serializer.serialize(value));
connection.close();
return success;
}
});
}
catch (Exception e)
{
logger.error("setNX redis error, key : {}", key);
}
return obj != null ? (Boolean) obj : false;
} private String getSet(final String key, final String value)
{
Object obj = null;
try
{
obj = redisTemplate.execute(new RedisCallback<Object>()
{
@Override
public Object doInRedis(RedisConnection connection) throws DataAccessException
{
StringRedisSerializer serializer = new StringRedisSerializer();
byte[] ret = connection.getSet(serializer.serialize(key), serializer.serialize(value));
connection.close();
return serializer.deserialize(ret);
}
});
}
catch (Exception e)
{
logger.error("setNX redis error, key : {}", key);
}
return obj != null ? (String) obj : null;
} /**
* 获得 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 = timeoutMsecs;
while (timeout >= 0)
{
long expires = System.currentTimeMillis() + expireMsecs + 1;
String expiresStr = String.valueOf(expires) + "&" + uuid; //锁到期时间 + uuid
if (this.setNX(lockKey, expiresStr))
{
// lock acquired
locked = true;
return true;
}
String currentValueStr = this.get(lockKey); //redis里的时间
if (currentValueStr != null && !"".equals(currentValueStr))
{
String[] split = currentValueStr.split("&");
if (split[0] != null && Long.parseLong(split[0]) < System.currentTimeMillis())
{
// 判断是否为空,不为空的情况下,如果被其他线程设置了值,则第二个条件判断是过不去的
// lock is expired
String oldValueStr = this.getSet(lockKey, expiresStr);
// 获取上一个锁到期时间,并设置现在的锁到期时间,
// 只有一个线程才能获取上一个线上的设置时间,因为jedis.getSet是同步的
if (oldValueStr != null && oldValueStr.equals(currentValueStr))
{
// 防止误删(覆盖,因为key是相同的)了他人的锁——这里达不到效果,这里值会被覆盖,但是因为什么相差了很少的时间,所以可以接受 // [分布式的情况下]:如过这个时候,多个线程恰好都到了这里,但是只有一个线程的设置值和当前值相同,他才有权利获取锁
// lock acquired
locked = true;
return true;
}
}
}
Thread.sleep(DEFAULT_ACQUIRY_RESOLUTION_MILLIS);
}
return false;
} /**
* Acqurired lock release.
*/
public synchronized void unlock()
{
if (locked)
{
String keyValue = this.get(lockKey);
if (keyValue != null && !"".equals(keyValue))
{
String[] split = keyValue.split("&");
// 判断uuid是否已经被修改
if (split.length == 2 && uuid.equals(split[1]))
{
redisTemplate.delete(lockKey);
}
}
locked = false;
}
} }

  

调用:

RedisLock lock = null;
try {
//先获取状态,如果为Y则任务已经发送完毕
if("Y".equals(redisutil.getString("flg"))){
return;
}
lock = new RedisLock(redisTemplate, "flg",10000,20000);//这里的flg最终会变成flg_lock
if(lock.lock()){
//再次检索,因为有可能开始获取状态不是Y,但是在获取锁的过程中被别人置成Y,所以再次检索确保不重发
if("Y".equals(redisutil.getString("flg"))){
return;
}
redisutil.set("flg", "Y",300);
    //======================================start
    //执行自己需要的任务
    //======================================end
  }
} catch (Exception e) {
e.printStackTrace();
}finally {
  if(lock != null)
  {lock.unlock();}
}

  redisutil网上应该有很多,应该能搜到

注意:虽然可以实现我想要的效果,但是我个人有个疑问,分布式系统的时间是怎样统一的,如果时间不一样,可能会出问题