公告:欢迎访问,查看更多资源请点我^.^!            点我关闭广告
分布式锁 Redis
2019-05-15 14:13:57
806人阅读
评论(0)
分类:分布式

    线上系统为分布式系统的时候 有一些业务逻辑是不是能并发执行的 需要在相同条件下 实现类似串行的状态 譬如:针对同一个用户的同一个接口操作。

    通过使用AOP结合Redis可以方便的实现分布式锁。

    首先编写redis的setNx方法(之后的redis版本会下线原有的setNx方法,所以使用set改写),使用set方法改装,set方法有很多参数:

String set(String key, String value, String nxxx, String expx, long time);

    nxxx: 只能取NX或者XX,如果取NX,则只有当key不存在是才进行set,如果取XX,则只有当key已经存在时才进行set;
    expx: 只能取EX或者PX,代表数据过期时间的单位,EX代表秒,PX代表毫秒;
    time: 过期时间,结合expx所代表的单位使用。

    详细:

    public boolean setNx(RedisDatabseEnum db, String key, String value, Integer ttl) {
if (db == null || key == null || key.trim().equals("")) {
logger.warn("参数不合法:db={},key={}", db == null ? "null" : db.getDatabaseId(), key);
throw new ParamInvalidException("参数不合法");
}
Jedis jedis = getJedisPool().getResource();
try {
// 指定database
jedis.select(db.getDatabaseId());
// 存入缓存
if (ttl != null && ttl > 0) {
String res = jedis.set(key, value, "nx", "px", ttl); // millis
return res != null && "ok".equalsIgnoreCase(res);
}
} catch (Exception e) {
logger.warn("setNx redis失败,key={},value={}", key,
(value != null && value.length() > 255) ? value.substring(0, 255) : value);
throw new BusinessException("setNx redis失败");
} finally {
if (jedis != null) {
jedis.close();
}
}
logger.debug("redis setNx success, db={}, key={} , value={} , ttl={}", db.getDatabaseId(), key,
(value != null && value.length() > 255) ? value.substring(0, 255) : value, ttl);

return false;
}

    然后编写注解:

import org.springframework.core.annotation.AliasFor;

import java.lang.annotation.*;

/**
* @author zhaochao
* @desc lock注解
*/
@Target({ElementType.PARAMETER, ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@Documented
public @interface Lock {

@AliasFor("key")
String value() default "";

/**
* 锁的key
*/
@AliasFor("value")
String key() default "";

/**
* 成功获取锁时value存储的时间 毫秒
*/
int saveMills() default 30 * 1000;

/**
* 如果未获取到锁,是否尝试重新获取锁
*/
boolean toWait() default true;

/**
* toWait为true时,重新获取锁的时间间隔 毫秒
*/
long sleepMills() default 50;

/**
* 最大尝试获取锁的次数 毫秒
*/
long maxTryTimes() default 10;

}

    编写主要逻辑AOP:

import com.x.x.exceptions.LockException;
import com.x.dao.sys.impl.RedisDaoImpl;
import com.x.enums.RedisDatabseEnum;
import org.apache.commons.lang.StringUtils;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.Signature;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.aspectj.lang.reflect.MethodSignature;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.LocalVariableTableParameterNameDiscoverer;
import org.springframework.expression.Expression;
import org.springframework.expression.ExpressionParser;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.stereotype.Component;

import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
* @author zhaochao
* @desc 切面
*/
@Aspect
@Component
public class LockAspect {
private final Logger logger = LoggerFactory.getLogger(this.getClass());

private final static String LOCK_PREFIX = "Lock::";
private ExpressionParser parser = new SpelExpressionParser();
private LocalVariableTableParameterNameDiscoverer discoverer = new LocalVariableTableParameterNameDiscoverer();

@Autowired
private RedisDaoImpl redis;

@Pointcut("@annotation(com.x.service.base.lock.Lock)")
public void lockAspect() {
}

@Around(value = "lockAspect()")
public Object lock(ProceedingJoinPoint pjp) throws Throwable {
Signature signature = pjp.getSignature();
String methodName = signature.getName();
Object[] args = pjp.getArgs();
logger.debug("LockAspect in lockAspect methodName = {}", methodName);

// 获取方法的注解
Method targetMethod = ((MethodSignature) signature).getMethod();
Method realMethod = pjp.getTarget().getClass().getDeclaredMethod(methodName, targetMethod.getParameterTypes());
YmtLock lock = realMethod.getAnnotation(YmtLock.class);
String key = "";
if (StringUtils.isNotBlank(lock.key())) {
Map<String, Object> params = new HashMap<>();
params.put("methodName", methodName);
params.put("fullName", targetMethod.getDeclaringClass().getName());
params.put("simpleName", targetMethod.getDeclaringClass().getSimpleName());
String[] paramList = discoverer.getParameterNames(targetMethod);
StandardEvaluationContext context = new StandardEvaluationContext();
context.setVariables(params);
for (int len = 0; len < paramList.length; len++) {
context.setVariable(paramList[len], args[len]);
}
logger.debug("LockAspect el key = {}", lock.key());
Expression expression = parser.parseExpression(lock.key());
key = expression.getValue(context, String.class);
}

key = StringUtils.isNotBlank(key) ? (LOCK_PREFIX + key) : (LOCK_PREFIX + targetMethod.getDeclaringClass().getName() + "." + methodName);

logger.debug("LockAspect key = {}", key);

boolean locked = false;
Object obj = null;
RedisDatabseEnum db = RedisDatabseEnum.TRACKING;
try {
// 如果未获取到锁,是否尝试重新获取锁
boolean toWait = lock.toWait();
// 重新获取锁的时间间隔
long sleepMills = lock.sleepMills();
// 最大尝试获取锁的次数
long maxTryTimes = lock.maxTryTimes();
// 成功获取锁时value存储的时间
int saveMills = lock.saveMills();
logger.debug("锁注解配置信息 toWait = {}, sleepMills = {}, maxTryTimes = {}, saveMills = {}", toWait, sleepMills, maxTryTimes, saveMills);
int tryTimes = 0;
while (!locked) {
++tryTimes;
// 尝试设置锁
locked = redis.setNx(db, key, String.valueOf(System.currentTimeMillis()), saveMills);
logger.debug("获取锁 次数 {} 结果 {}", tryTimes, locked);
// 设置成功 执行业务
if (locked) {
obj = pjp.proceed();
}
// 需要重试
else if (toWait) {
// 超过最大重试次数
if (tryTimes >= maxTryTimes) {
logger.debug("经过 {} 次尝试仍未获取到该业务锁", maxTryTimes);
throw new LockException("经过" + maxTryTimes + "次尝试仍未获取到该业务锁,请稍后重试");
}
TimeUnit.MILLISECONDS.sleep(sleepMills);
} else {
logger.debug("其他线程正在执行该业务");
throw new LockException("其他线程正在执行该业务,请稍后重试");
}
}
} catch (Throwable e) {
throw e;
} finally {
// 如果获取到了锁,释放锁
if (locked) {
redis.clear(db, key);
}
}
return obj;
}

}

    

    这样在需要的service方法上添加注解即可,实现了多次重试机制;需要的话可以改写为最大重试时长,其实是一样的。

分享一下:
赞一下(4)
博主资料
博主头像
zc521106
文章:34
浏览:19899
文章分类
Java(12)
NoSql(112)
数据库(12)
前端(12)
阅读排行
Java基础知识
(12)
NoSql应用
(112)
数据库Oracle语法
(12)
前端常用工具类
(12)
java web项目
(12)
linux安装mysql
(12)
评论区
这篇文章怎么样?写点评论吧!
姓名:
邮箱:
有回复时通知我:
发表
回复【10.20.30.32】:
暂无评论,快抢沙发!
* 以上用户言论只代表其个人观点,不代表iBlog网站的观点或立场,如有任何疑问请随时联系管理员...