最近抽空优化了之前已有的redis分布式锁，主要用于解决高并发的问题，比如抢红包，多个人同时操作红包库存，当在库存只剩下1个的时候，一个人的减库存的操作事务没提交，另一个人的查库存操作刚好同步执行，这样就会出现很尴尬的事情，1个红包会被2个人抢走，这个时候，我们就要依托锁，将请求入口锁住，当然锁有很多种方式，这边就记录一下比较好用的redis分布式锁。
方式有很多setNX 、set、incr等等，setNX只要通过逻辑防止死锁就可以了
直接上代码：
public boolean keyLock(final String key, final long keepMin) { boolean obj = false; try { obj = (boolean) redisTemplateSerializable.execute(new RedisCallback<Object>() { @Override public Object doInRedis(RedisConnection connection) throws DataAccessException { try{ Long incr = connection.incr(key.getBytes()); if(incr == 1){ connection.setEx(key.getBytes(), keepMin, incr.toString().getBytes()); return true; }else{ Long ttl = connection.ttl(key.getBytes()); if(ttl == -1){ //设置失败,重新设置过期时间 connection.setEx(key.getBytes(), keepMin, incr.toString().getBytes()); return true; } } }catch (Exception e) { logger.error("加锁异常", e); connection.del(key.getBytes()); return true; } return false; } }); }catch (Exception e) { logger.error(e.getMessage()); } return obj; }
注解
package com.tp.soft.common.interceptor; import java.lang.annotation.Documented; import java.lang.annotation.ElementType; import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; import java.util.concurrent.TimeUnit; /** * redis锁注解 * * @author taop */ @Retention(RetentionPolicy.RUNTIME) @Target({ ElementType.METHOD }) @Documented public @interface RedisLock { String lockName() default ""; // 锁名 int retryTimes() default 0; // 重试次数 long retryWait() default 200; // 重试等待时间,单位 : ms int keeyMinTime() default 1; //锁自动失效时间 1秒 }
aop
package com.tp.soft.aop.redis; import java.lang.reflect.Method; import java.util.HashMap; import java.util.Map; import javax.annotation.Resource; 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.core.LocalVariableTableParameterNameDiscoverer; 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 cn.hutool.core.lang.Assert; import com.tp.soft.common.interceptor.Cacheable; import com.tp.soft.common.interceptor.RedisLock; import com.tp.soft.redis.RedisCacheSvc; @Aspect @Component public class RedisLockAop { private static final Logger log = LoggerFactory.getLogger(RedisLockAop.class); private static final String LOCK_NAME = "lockName"; private static final String RETRY_TIMES = "retryTimes"; private static final String RETRY_WAIT = "retryWait"; private static final String KEEP_MIN_TIME = "keepMinTime"; @Resource private RedisCacheSvc redisCacheSvc; @Pointcut("@annotation(com.tp.soft.common.interceptor.RedisLock)") public void redisLockAspect() { } @Around("redisLockAspect()") public Object lockAroundAction(ProceedingJoinPoint pjp) throws Throwable { Method method = returnMethod(pjp); Map<String, Object> annotationArgs = this.getAnnotationArgs(pjp); String lockPrefix = (String) annotationArgs.get(LOCK_NAME); Assert.notNull(lockPrefix, "分布式,锁名不能为空"); int retryTimes = (int) annotationArgs.get(RETRY_TIMES); long retryWait = (long) annotationArgs.get(RETRY_WAIT); int keepMinTime = (int) annotationArgs.get(KEEP_MIN_TIME); String keyName = parseKey(lockPrefix, method, pjp.getArgs()); // 获取redis锁,防止死锁 boolean keyLock = redisCacheSvc.keyLock(keyName, keepMinTime); if(keyLock){ //执行主程序 return pjp.proceed(); }else{ if(retryTimes <= 0){ log.info(String.format("{%s}已经被锁, 不重试", keyName)); throw new RuntimeException(String.format("{%s}已经被锁, 不重试", keyName)); } int failCount = 1; while (failCount <= retryTimes) { // 等待指定时间ms try { Thread.sleep(retryWait); } catch (InterruptedException e) { e.printStackTrace(); } if (redisCacheSvc.keyLock(keyName, keepMinTime)) { // 执行主逻辑 return pjp.proceed(); } else { log.info(String.format("{%s}已经被锁, 正在重试[ %s/%s ],重试间隔{%s}毫秒", keyName, failCount, retryTimes, retryWait)); failCount++; } } throw new RuntimeException("系统繁忙, 请稍等再试"); } } /** * 获取锁参数 * * @param proceeding * @return */ private Map<String, Object> getAnnotationArgs(ProceedingJoinPoint proceeding) { Class target = proceeding.getTarget().getClass(); Method[] methods = target.getMethods(); String methodName = proceeding.getSignature().getName(); for (Method method : methods) { if (method.getName().equals(methodName)) { Map<String, Object> result = new HashMap<String, Object>(); RedisLock redisLock = method.getAnnotation(RedisLock.class); result.put(LOCK_NAME, redisLock.lockName()); result.put(RETRY_TIMES, redisLock.retryTimes()); result.put(RETRY_WAIT, redisLock.retryWait()); result.put(KEEP_MIN_TIME, redisLock.keeyMinTime()); return result; } } return null; } private Method returnMethod(ProceedingJoinPoint pjp) throws NoSuchMethodException { Signature signature = pjp.getSignature(); Class<? extends Object> cls = pjp.getTarget().getClass(); MethodSignature methodSignature = (MethodSignature) signature; Method targetMethod = methodSignature.getMethod(); Method method = cls.getDeclaredMethod(signature.getName(), targetMethod.getParameterTypes()); return method; } /** * 获取缓存的key key 定义在注解上，支持SPEL表达式 * * @param pjp * @return */ private String parseKey(String key, Method method, Object[] args) { // 获取被拦截方法参数名列表(使用Spring支持类库) LocalVariableTableParameterNameDiscoverer u = new LocalVariableTableParameterNameDiscoverer(); String[] paraNameArr = u.getParameterNames(method); // 使用SPEL进行key的解析 ExpressionParser parser = new SpelExpressionParser(); // SPEL上下文 StandardEvaluationContext context = new StandardEvaluationContext(); // 把方法参数放入SPEL上下文中 for (int i = 0; i < paraNameArr.length; i++) { context.setVariable(paraNameArr[i], args[i]); } return parser.parseExpression(key).getValue(context, String.class); } }
搭建完成后直接在需要锁住的接口上注解
@RedisLock(lockName="'lock_'+#tbbId",retryTimes=5)
模拟高并发测试
for (int i = 0; i < 2; i++) { threadPoolTaskExecutor.execute(new StartTaskThread(redisCacheSvc, i, threadPoolTaskExecutor)); }
效果就是这样了
菜鸟的思路，各位大神有好的思路请留言…
