redis 分布式锁

最近抽空优化了之前已有的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));
        }

效果就是这样了

菜鸟的思路，各位大神有好的思路请留言...