redis分布式锁，处理/user/center/save/userExtInfo并发问题，导致MySQLIntegrityConstraintViolationException

pom.xml

@@ -183,6 +183,11 @@
             <artifactId>commons-collections</artifactId>
             <version>3.2.1</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-text</artifactId>
+            <version>1.1</version>
+        </dependency>
 
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>

src/main/java/cn/quantgroup/xyqb/controller/ExceptionHandlingController.java

@@ -3,6 +3,7 @@ package cn.quantgroup.xyqb.controller;
 import cn.quantgroup.xyqb.exception.*;
 import cn.quantgroup.xyqb.model.JsonResult;
 import cn.quantgroup.xyqb.util.IpUtil;
+import cn.quantgroup.xyqb.util.lock.redislock.ResubmissionException;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.http.HttpStatus;
 import org.springframework.jdbc.BadSqlGrammarException;
@@ -165,4 +166,11 @@ public class ExceptionHandlingController implements IBaseController {
         log.error("sql语法解析异常 error sql = 【{}】", e.getSql(), e);
         return JsonResult.buildErrorStateResult("参数错误。", null);
     }
+
+    @ExceptionHandler(ResubmissionException.class)
+    @ResponseStatus(HttpStatus.OK)
+    @ResponseBody
+    public JsonResult resubmissionException(ResubmissionException re) {
+        return new JsonResult(re.getMessage(), 0L, null, re.getBusinessCode());
+    }
 }

src/main/java/cn/quantgroup/xyqb/controller/internal/user/center/UserCenterController.java

@@ -9,6 +9,7 @@ import cn.quantgroup.xyqb.aspect.limit.PasswordFreeAccessValidator;
 import cn.quantgroup.xyqb.entity.*;
 import cn.quantgroup.xyqb.model.JsonResult;
 import cn.quantgroup.xyqb.service.user.*;
+import cn.quantgroup.xyqb.util.lock.redislock.RedisLock;
 import com.alibaba.fastjson.JSONObject;
 import com.alibaba.fastjson.TypeReference;
 import io.swagger.annotations.ApiOperation;
@@ -324,6 +325,7 @@ public class UserCenterController {
      * @return
      */
     @RequestMapping("/save/userExtInfo")
+    @RedisLock(prefix = "lock:user:ext:", key = "#this[0]")
     @ApiOperation(value = "保存用户经济学历等信息", notes = "保存用户经济学历等信息", httpMethod = "POST")
     public JsonResult saveUserExtInfo(String phoneNo, EducationEnum educationEnum, MaritalStatus maritalStatus, IncomeRangeEnum incomeRangeEnum, OccupationEnum occupationEnum) {
         if (StringUtils.isEmpty(phoneNo)) {

src/main/java/cn/quantgroup/xyqb/util/lock/RedisLock.java

+package cn.quantgroup.xyqb.util.lock;
+
+import org.apache.commons.text.RandomStringGenerator;
+import org.springframework.dao.InvalidDataAccessResourceUsageException;
+import org.springframework.data.redis.core.RedisCallback;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.serializer.StringRedisSerializer;
+
+import java.util.Random;
+
+/**
+ * RedisLock
+ * Created by Feng on 2017/6/5.
+ * <pre>ex:
+ * try{
+ * if(redisLock.lock()){
+ * // CODE
+ * }
+ * } catch (InterruptedException e) {
+ * LOGGER.warn("获取锁失败：lockKey:{},exception:{}",redisLock.getLockKey(),e.getMessage());
+ * }finally {
+ * redisLock.unlock();
+ * }
+ * </pre>
+ */
+public class RedisLock {
+    private final static org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(RedisLock.class);
+    private RedisTemplate redisTemplate;
+    /**
+     * Lock key path.
+     */
+    private String lockKey;
+
+    /**
+     * 锁超时时间，防止线程在入锁以后，无限的执行等待
+     */
+    private int expireMsecs = 60 * 1000;
+
+    /**
+     * 锁等待时间，防止线程饥饿
+     */
+    private int timeoutMsecs = 10 * 1000;
+
+    /**
+     * 当前锁的到期时间字符串
+     */
+    private volatile String expiresStr = null;
+
+
+    /**
+     * Detailed constructor with default acquire timeout 10000 ms and lock expiration of 60000 ms.
+     *
+     * @param redisTemplate 注入一个 redisTemplate
+     * @param lockKey       lock key (ex. account:1, ...)
+     */
+    public RedisLock(RedisTemplate redisTemplate, String lockKey) {
+        this.redisTemplate = redisTemplate;
+        this.lockKey = "lock:".concat(lockKey).concat("_lock");
+        logger.info("准备锁 lock:{},{}", getLockKey());
+    }
+
+    /**
+     * Detailed constructor with default lock expiration of 60000 ms.
+     *
+     * @param redisTemplate 注入一个 redisTemplate
+     * @param lockKey       锁KEY  (ex. account:1, ...)
+     * @param timeoutMsecs  锁等待时间 默认  10 * 1000 ms
+     */
+    public RedisLock(RedisTemplate redisTemplate, String lockKey, int timeoutMsecs) {
+        this(redisTemplate, lockKey);
+        this.timeoutMsecs = timeoutMsecs;
+    }
+
+    /**
+     * Detailed constructor.
+     *
+     * @param redisTemplate 注入一个 redisTemplate
+     * @param lockKey       锁KEY  (ex. account:1, ...)
+     * @param timeoutMsecs  锁等待时间 默认  10 * 1000 ms
+     * @param expireMsecs   锁超时时间 默认 60 * 1000 ms
+     */
+    public RedisLock(RedisTemplate redisTemplate, String lockKey, int timeoutMsecs, int expireMsecs) {
+        this(redisTemplate, lockKey, timeoutMsecs);
+        this.expireMsecs = expireMsecs;
+    }
+
+    public String getLockKey() {
+        return lockKey;
+    }
+
+    /**
+     * 获取key -> value
+     *
+     * @param key
+     * @return
+     */
+    private String get(final String key){
+        Object obj = null;
+        try {
+            obj = redisTemplate.execute((RedisCallback<Object>) connection -> {
+                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 + ": {}", e);
+            throw new InvalidDataAccessResourceUsageException(e.getMessage());
+        }
+        return obj != null ? obj.toString() : null;
+    }
+
+    /**
+     * SET if Not exists
+     *
+     * @param key
+     * @param value
+     * @return
+     */
+    private boolean setNX(final String key, final String value) {
+        Object obj = null;
+        try {
+            obj = redisTemplate.execute((RedisCallback<Object>) connection -> {
+                StringRedisSerializer serializer = new StringRedisSerializer();
+                byte[] skey = serializer.serialize(key);
+                Boolean success = connection.setNX(skey, serializer.serialize(value));
+                int expire = expireMsecs / 1000;
+                expire = expire <= 0 ? 120 : expire + 1;
+                connection.expire(skey, expire);
+                connection.close();
+                return success;
+            });
+        } catch (Exception e) {
+            logger.error("setNX redis error, key “" + key + "”: {}", e);
+        }
+        return obj != null ? (Boolean) obj : false;
+    }
+
+    /**
+     * getSet 命令在Redis键中设置指定的字符串值，并返回其旧值
+     *
+     * @param key
+     * @param value
+     * @return
+     */
+    private String getSet(final String key, final String value) {
+        Object obj = null;
+        try {
+            obj = redisTemplate.execute((RedisCallback<Object>) connection -> {
+                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("getSet redis error, key : {}", key);
+        }
+        return obj != null ? (String) obj : null;
+    }
+
+    /**
+     * lock
+     *
+     * @return
+     * @throws InterruptedException
+     */
+    public synchronized boolean lock() throws InterruptedException {
+        int timeout = timeoutMsecs;
+        int retryCount = 0;
+        while (timeout >= 0) {
+            long expires = System.currentTimeMillis() + expireMsecs + 1;
+            RandomStringGenerator generator3 = new RandomStringGenerator.Builder()
+                    .withinRange('0', 'z').build();
+            String random = generator3.generate(5);
+            String expiresStr = random + String.valueOf(expires);
+            if (this.setNX(lockKey, expiresStr)) {
+                // lock acquired
+                this.expiresStr = expiresStr;
+                logger.info("获取锁成功 lock:{}", getLockKey());
+                return true;
+            }
+
+            String currentValueStr; //redis里的时间
+            try {
+                if ((currentValueStr = this.get(lockKey)) == null) {
+                    continue;
+                }
+                if (Long.parseLong(currentValueStr.substring(5)) < System.currentTimeMillis()) {
+                    //判断是否为空，不为空的情况下，如果被其他线程设置了值，则第二个条件判断是过不去的
+                    // lock is expired
+                    String oldValueStr = this.getSet(lockKey, expiresStr);
+                    //获取上一个锁到期时间，并设置现在的锁到期时间，
+                    //只有一个线程才能获取上一个线上的设置时间，因为jedis.getSet是同步的
+                    if (currentValueStr.equals(oldValueStr)) {
+                        //防止误删（覆盖，因为key是相同的）了他人的锁——这里达不到效果，这里值会被覆盖，但是因为什么相差了很少的时间，所以可以接受
+                        //[分布式的情况下]:如过这个时候，多个线程恰好都到了这里，但是只有一个线程的设置值和当前值相同，他才有权利获取锁
+                        // lock acquired
+                        this.expiresStr = expiresStr;
+                        logger.info("获取锁成功（前一个锁超时） lock:{}", getLockKey());
+                        return true;
+                    }
+                }
+                int sleepTime = new Random().nextInt(10) * 10;
+                sleepTime += 100; //保证随机等待时间为 100-200ms
+                timeout -= sleepTime;
+                /*
+                    延迟 随机的等待时间 毫秒, 防止饥饿进程的出现,即,当同时到达多个进程,
+                    只会有一个进程获得锁,其他的都用随机的频率进行尝试,后面有来了一些进行,也以同样的频率申请锁,这将可能导致前面来的锁得不到满足.
+                 */
+                Thread.sleep(sleepTime);
+            } catch (InvalidDataAccessResourceUsageException e) {
+                if (retryCount > 3) {
+                    logger.error("redis有可能出现问题不能获取lockKey：{}", lockKey);
+                    logger.warn("redis lock 3次重试仍异常，lockKey: {} 进行降级服务....", lockKey);
+                    return true;
+                }
+                retryCount++;
+                if (timeout > 100) {
+                    timeout -= 100; }
+                else
+                    timeout = 1; {
+                Thread.sleep(100); }
+            }
+        }
+        logger.info("锁失败 lock:{}", getLockKey());
+        return false;
+    }
+
+    /**
+     * 解锁
+     */
+    public synchronized void unlock() {
+        if (this.expiresStr != null) {
+            /*如果存在超时字符串则 判断是否是自己的超时字符串*/
+            try {
+                String expStr = null;
+                try {
+                    expStr = this.get(lockKey);
+                } catch (InvalidDataAccessResourceUsageException e) {
+                    logger.warn("redis unlock 异常， lockKey:{} 进行降级服务....", lockKey);
+                }
+                if (expStr == null || this.expiresStr.equals(expStr)) {
+                    redisTemplate.delete(lockKey);
+                    logger.info("解锁成功 lock:{}", getLockKey());
+                    return;
+                }
+            } catch (Exception e) {
+                logger.error("redis unlock 异常 lockKey "+lockKey+"：{} ", e);
+            }
+        }
+        logger.warn("没有锁或者因非本对象产生的锁，不能进行解锁，因为是不安全的");
+
+    }
+
+}

src/main/java/cn/quantgroup/xyqb/util/lock/redislock/RedisLock.java

+package cn.quantgroup.xyqb.util.lock.redislock;
+
+import java.lang.annotation.*;
+
+/**
+ * 用redis进行锁<br/>
+ * 如果上一个请求正在处理会抛出throw ResubmissionException<br/>
+ * Created by Feng on 2017/6/6.
+ */
+@Documented
+@Target(ElementType.METHOD)
+@Retention(RetentionPolicy.RUNTIME)
+public @interface RedisLock {
+
+    /**
+     * 指定 lock key ;指定value后， key\prefix 设置将无效；
+     * 默认为：方法名+":"+参数hash
+     * @return
+     */
+    String value() default "";
+
+    /**
+     * 指定业务 key ;  默认为参数hash；可以是 SPEL表达式从参数中取值:(ex: #this[0].id - 第一个参数ID... )
+     * ;最终lock key 为 prefix +":"+ key
+     * @return
+     */
+    String key() default "";
+
+    /**
+     * 指定前缀  默认为 方法名；最终lock key 为 prefix +":"+ key
+     * @return
+     */
+    String prefix() default "";
+
+    /**
+     * 锁等待时间 默认  10 * 1000 ms
+     * @return
+     */
+    int timeout() default 10 * 1000;//
+
+    /**
+     * 锁超时时间 默认 60 * 1000 ms
+     * @return
+     */
+    int expire() default 60 * 1000;//
+
+}

src/main/java/cn/quantgroup/xyqb/util/lock/redislock/RedisLockAspect.java

+package cn.quantgroup.xyqb.util.lock.redislock;
+
+import com.alibaba.fastjson.JSONObject;
+import org.apache.commons.codec.digest.DigestUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.aspectj.lang.ProceedingJoinPoint;
+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.beans.factory.annotation.Qualifier;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.expression.Expression;
+import org.springframework.expression.ExpressionException;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.stereotype.Component;
+
+import java.lang.reflect.Method;
+
+/**
+ * Redis Lock Aspect
+ */
+@Aspect
+@Component
+public class RedisLockAspect {
+
+    private final static Logger LOGGER = LoggerFactory.getLogger(RedisLockAspect.class);
+
+    @Autowired
+    @Qualifier("stringRedisTemplate")
+    private RedisTemplate<String, String> stringRedisTemplate;
+
+    @Pointcut("@annotation(cn.quantgroup.xyqb.util.lock.redislock.RedisLock)")
+    private void redisLockPointCut() {
+    }
+
+    /**
+     * @param pjp
+     * @return
+     * @throws Throwable ResubmissionException 上一个请求正在处理
+     */
+    @Around("redisLockPointCut()")
+    private Object preventDuplicateSubmit(ProceedingJoinPoint pjp) throws Throwable {
+        MethodSignature methodSignature = (MethodSignature) pjp.getSignature();
+        Method method = methodSignature.getMethod();
+        RedisLock annotation = method.getAnnotation(RedisLock.class);
+        Object[] args = pjp.getArgs();
+        String lockKey = null;
+        int timeout = annotation.timeout();
+        int expire = annotation.expire();
+        String prefix = annotation.prefix();
+        if (StringUtils.isBlank(prefix)) {
+            prefix = method.getName();
+        }
+        if (StringUtils.isNotBlank(annotation.value())) {
+            //指定了lockKey
+            lockKey = annotation.value();
+        } else if (StringUtils.isNotBlank(annotation.key())) {
+            String keySPEL = annotation.key();
+            try {
+                if (keySPEL.startsWith("#this")) {//判断是否是spel表达式
+                    Expression expression = new SpelExpressionParser().parseExpression(keySPEL);
+                    String value = expression.getValue(args, String.class);
+                    lockKey = prefix.concat(":").concat(value);
+                } else {
+                    lockKey = prefix.concat(":").concat(keySPEL);
+                }
+            } catch (ExpressionException e) {
+                LOGGER.error("key表达式“" + keySPEL + "”错误：{}", e);
+                throw e;
+            }
+        } else {
+            if (args != null && args.length > 1) {
+                lockKey = prefix.concat(":").concat(DigestUtils.md5Hex(JSONObject.toJSONString(args)));
+            } else {
+                return pjp.proceed();
+            }
+        }
+        cn.quantgroup.xyqb.util.lock.RedisLock lock = new cn.quantgroup.xyqb.util.lock.RedisLock(stringRedisTemplate, lockKey, timeout, expire);
+        try {
+            if (lock.lock()) {
+                return pjp.proceed();
+            } else {
+                LOGGER.warn("调用方法失败，已有业务数据在处理中 lockKey：{}", lock.getLockKey());
+                throw new ResubmissionException();
+            }
+        } catch (InterruptedException e) {
+            LOGGER.warn("获取锁失败：lockKey:{},exception:{}", lock.getLockKey(), e.getMessage());
+            throw new ResubmissionException();
+        } finally {
+            lock.unlock();
+        }
+    }
+}

src/main/java/cn/quantgroup/xyqb/util/lock/redislock/ResubmissionException.java

+package cn.quantgroup.xyqb.util.lock.redislock;
+
+/**
+ * @author Jie.Feng
+ * @date 2017/12/14
+ */
+public class ResubmissionException extends RuntimeException {
+
+    private Long businessCode = 3L;
+
+    public ResubmissionException() {
+        super("你操作太快了,请歇息一下");
+    }
+
+    public ResubmissionException(String msg) {
+        super(msg);
+    }
+
+    public ResubmissionException(Long businessCode, String msg) {
+        super(msg);
+        this.businessCode = businessCode;
+    }
+
+    public Long getBusinessCode() {
+        return businessCode;
+    }
+}