Java 分布式服务重复提交解决方案 Redis

本文实现一种分布式服务防重复提交的方案也就是一线一个锁，在方法请求前，要先获取锁如果锁存在则返回异常。

下面简单介绍一下如何使用Redis实现分布式锁

CacheLock.java 为自定义注解接口，CacheLock方法注解用来指定分布式锁的key前缀和失效时间等信息
LockKeyGenerator.java为切面，用于拦截Heders中token参数，生成分布式锁的key
LockMethodInterceptor.java为切面，用于拦截@CacheLock方法，实现在执行方法之前要先获取锁逻辑
RedisLockHelper.java为分布式锁的实现

1.pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.service</groupId><artifactId>springboot-repeat-submit</artifactId><version>1.0-SNAPSHOT</version><!-- Spring Boot 启动父依赖 --><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.0.3.RELEASE</version></parent><dependencies><!-- Exclude Spring Boot's Default Logging --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-aop</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId></dependency><!-- https://mvnrepository.com/artifact/redis.clients/jedis --><dependency><groupId>redis.clients</groupId><artifactId>jedis</artifactId><version>2.9.0</version></dependency><!-- https://mvnrepository.com/artifact/org.projectlombok/lombok --><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>1.18.2</version><scope>provided</scope></dependency><dependency><groupId>com.google.guava</groupId><artifactId>guava</artifactId><version>21.0</version></dependency></dependencies></project>

2.自定义注解

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
public @interface CacheLock {/*** redis 锁key的前缀*/String prefix() default "";/*** redis key过期时间*/int expire() default 5;/*** 超时时间单位**/TimeUnit timeUnit() default TimeUnit.SECONDS;/*** Key分隔符* 比如：Key:1*/String delimiter() default ":";
}

3.分布式锁key 生成

public class LockKeyGenerator implements CacheKeyGenerator {@Overridepublic String getLockKey(ProceedingJoinPoint pjp) {MethodSignature signature = (MethodSignature) pjp.getSignature();Method method = signature.getMethod();CacheLock lockAnnotation = method.getAnnotation(CacheLock.class);StringBuilder builder = new StringBuilder();HttpServletRequest request = ((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getRequest();String token = request.getHeader(HttpHeaders.AUTHORIZATION);builder.append(lockAnnotation.delimiter()).append(token);return lockAnnotation.prefix() + builder.toString();}
}

如果想要使用请求参数中的属性生成分布式key
可以添加自定义注解 CacheParam.java 使用@CacheParam参数生产key
@CacheParam需要作用在请求参数上

@Target({ElementType.PARAMETER, ElementType.METHOD, ElementType.FIELD})
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
public @interface CacheParam {/*** 字段名称** @return String*/String name() default "";
}

public class LockKeyGenerator implements CacheKeyGenerator {@Overridepublic String getLockKey(ProceedingJoinPoint pjp) {MethodSignature signature = (MethodSignature) pjp.getSignature();Method method = signature.getMethod();CacheLock lockAnnotation = method.getAnnotation(CacheLock.class);final Object[] args = pjp.getArgs();final Parameter[] parameters = method.getParameters();StringBuilder builder = new StringBuilder();//默认解析方法里面带CacheParam注解的属性,如果没有尝试着解析实体对象中的CacheParam注解属性for (int i = 0; i < parameters.length; i++) {final CacheParam annotation = parameters[i].getAnnotation(CacheParam.class);if (annotation == null) {continue;}builder.append(lockAnnotation.delimiter()).append(args[i]);}if (StringUtils.isEmpty(builder.toString())) {//CacheLock注解的方法参数没有CacheParam注解，则迭代解析参数实体中的CacheParam注解属性final Annotation[][] parameterAnnotations = method.getParameterAnnotations();for (int i = 0; i < parameterAnnotations.length; i++) {final Object object = args[i];final Field[] fields = object.getClass().getDeclaredFields();for (Field field : fields) {final CacheParam annotation = field.getAnnotation(CacheParam.class);if (annotation == null) {continue;}field.setAccessible(true);builder.append(lockAnnotation.delimiter()).append(ReflectionUtils.getField(field, object));}}}return lockAnnotation.prefix() + builder.toString();}
}

集成token
如果没有token 方法上添加@RequestParam @RequestBody注解可在方法上添加@CacheParam(name=“对应的字段名”)

public class LockKeyGenerator implements CacheKeyGenerator {@Overridepublic String getLockKey(ProceedingJoinPoint pjp) {CacheLock lockAnnotation = null;StringBuilder builder = null;try {MethodSignature signature = (MethodSignature) pjp.getSignature();Method method = signature.getMethod();lockAnnotation = method.getAnnotation(CacheLock.class);PassLogin passLoginAnnotation = method.getAnnotation(PassLogin.class);builder = new StringBuilder();if(passLoginAnnotation == null){HttpServletRequest request = ((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getRequest();String token = request.getHeader(HttpHeaders.AUTHORIZATION);builder.append(lockAnnotation.delimiter()).append(token);} else {final Object[] args = pjp.getArgs();final Parameter[] parameters = method.getParameters();//默认解析方法里面带CacheParam注解的属性,如果没有尝试着解析实体对象中的CacheParam注解属性for (int i = 0; i < parameters.length; i++) {final CacheParam annotation = parameters[i].getAnnotation(CacheParam.class);if (annotation == null) {continue;} else {final Object object = args[i];if(object instanceof String){builder.append(lockAnnotation.delimiter()).append(args[i]);} else {final Field[] fields = object.getClass().getDeclaredFields();if(fields != null){for (Field field : fields) {if(field.getName().equals(annotation.name())){field.setAccessible(true);builder.append(lockAnnotation.delimiter()).append(ReflectionUtils.getField(field, object));break;}}}}}}if (StringUtils.isEmpty(builder.toString())) {//CacheLock注解的方法参数没有CacheParam注解，则迭代解析参数实体中的CacheParam注解属性final Annotation[][] parameterAnnotations = method.getParameterAnnotations();for (int i = 0; i < parameterAnnotations.length; i++) {final Object object = args[i];final Field[] fields = object.getClass().getDeclaredFields();for (Field field : fields) {final CacheParam annotation = field.getAnnotation(CacheParam.class);if (annotation == null) {continue;}field.setAccessible(true);builder.append(lockAnnotation.delimiter()).append(ReflectionUtils.getField(field, object));}}}}} catch (Exception e) {e.printStackTrace();}return lockAnnotation.prefix() + builder.toString();}

4.分布式锁实现

@Configuration
public class RedisConfig {@Beanpublic RedisConnectionFactory redisConnectionFactory() {RedisStandaloneConfiguration redisStandaloneConfiguration = new RedisStandaloneConfiguration("127.0.0.1", 6379);return new JedisConnectionFactory(redisStandaloneConfiguration);}@Beanpublic RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {RedisTemplate<String, Object> template = new RedisTemplate<>();template.setConnectionFactory(redisConnectionFactory);template.setKeySerializer(new StringRedisSerializer());template.setValueSerializer(new GenericJackson2JsonRedisSerializer());return template;}}

@Configuration
@RequiredArgsConstructor
public class RedisLockHelper {private static final String DELIMITER = "|";private static final ScheduledExecutorService EXECUTOR_SERVICE = Executors.newScheduledThreadPool(10);private final StringRedisTemplate stringRedisTemplate;/*** 获取锁* @param lockKey lockKey* @param uuid    UUID* @param timeout 超时时间* @param unit    过期单位* @return true or false*/public boolean lock(String lockKey, final String uuid, long timeout, final TimeUnit unit) {final long milliseconds = Expiration.from(timeout, unit).getExpirationTimeInMilliseconds();boolean success = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, (System.currentTimeMillis() + milliseconds) + DELIMITER + uuid);if (success) {/*设置过期时间，防止系统崩溃而导致锁迟迟不释放形成死锁*/stringRedisTemplate.expire(lockKey, timeout, unit);} else {String oldVal = stringRedisTemplate.opsForValue().get(lockKey);final String[] oldValues = oldVal.split(Pattern.quote(DELIMITER));/*缓存已经到过期时间，但是还没释放，避免ddl失效造成死锁*/if (Long.parseLong(oldValues[0]) + unit.toSeconds(1) <= System.currentTimeMillis()) {stringRedisTemplate.opsForValue().set(lockKey, (System.currentTimeMillis() + milliseconds) + DELIMITER + uuid);stringRedisTemplate.expire(lockKey, timeout, unit);return true;}}return success;}public void unlock(String lockKey, String value) {unlock(lockKey, value, 0, TimeUnit.MILLISECONDS);}/*** 延迟unlock** @param lockKey   key* @param uuid* @param delayTime 延迟时间* @param unit      时间单位*/private void unlock(final String lockKey, final String uuid, long delayTime, TimeUnit unit) {if (StringUtils.isEmpty(lockKey)) {return;}if (delayTime <= 0) {doUnlock(lockKey, uuid);} else {/*定时任务延迟unlock*/EXECUTOR_SERVICE.schedule(() -> doUnlock(lockKey, uuid), delayTime, unit);}}/*** @param lockKey key* @param uuid*/private void doUnlock(final String lockKey, final String uuid) {String val = stringRedisTemplate.opsForValue().get(lockKey);final String[] values = val.split(Pattern.quote(DELIMITER));if (values.length <= 0) {return;}if (uuid.equals(values[1])) {stringRedisTemplate.delete(lockKey);}}}

简单讲一下锁的实现，Redis是线程安全的，利用该的特性可以很轻松的实现一个分布式锁。opsForValue().setIfAbsent(key,value)的作用是如果缓存中没有当前Key则进行缓存同时返回true，否则返回false。只靠这一个逻辑其实也算是实现了锁，但是为了防止防止系统崩溃而导致锁迟迟不释放形成死锁，或者Redis ddl失效导致死锁，又添加一些比如key失效时间等逻辑。可以仔细读一下，并不难理解。

5.分布式锁切面

拦截@CacheLock注解方法，在方法执行前增加获取锁逻辑

@Aspect
@Configuration
@AllArgsConstructor
public class LockMethodInterceptor {private final RedisLockHelper redisLockHelper;private final CacheKeyGenerator cacheKeyGenerator;@Around("execution(public * *(..)) && @annotation(com.zhuoli.service.springboot.distributed.repeat.submit.annotation.CacheLock)")public Object interceptor(ProceedingJoinPoint pjp) {MethodSignature signature = (MethodSignature) pjp.getSignature();Method method = signature.getMethod();CacheLock lock = method.getAnnotation(CacheLock.class);if (StringUtils.isEmpty(lock.prefix())) {throw new RuntimeException("lock key don't null...");}final String lockKey = cacheKeyGenerator.getLockKey(pjp);String value = UUID.randomUUID().toString();try {final boolean success = redisLockHelper.lock(lockKey, value, lock.expire(), lock.timeUnit());if (!success) {throw new RuntimeException("重复提交");}try {return pjp.proceed();} catch (Throwable throwable) {throw new RuntimeException("系统异常");}} finally {//如果演示的话需要注释该代码，实际应该放开redisLockHelper.unlock(lockKey, value);}}
}

6.注解使用

该方法使用token生成key
key = prefix + “:” + token

    @CacheLock(prefix = "checkDeviceNo")@RequestMapping(value = "/checkDeviceNo", method = {RequestMethod.POST})public R checkDeviceNo(@RequestParam String deviceNo) {Integer countDeviceNo = cameraDeviceService.countDeviceNo(deviceNo);if (countDeviceNo != null && countDeviceNo > 0) {return R.failed(ErrorCodeEnum.PB10010001.msg());} else {return R.ok();}}

使用请求参数生成key
key = prefix + “:” + deviceNo

    @CacheLock(prefix = "checkDeviceNo")@RequestMapping(value = "/checkDeviceNo", method = {RequestMethod.POST})public R checkDeviceNo(@CacheParam(name = "deviceNo") @RequestParam String deviceNo) {Integer countDeviceNo = cameraDeviceService.countDeviceNo(deviceNo);if (countDeviceNo != null && countDeviceNo > 0) {return R.failed(ErrorCodeEnum.PB10010001.msg());} else {return R.ok();}}

以上就是实现Java分布式锁重复提交问题的解决方案，欢迎提更好的解决方案！！！