SpringBoot 集成 redis 分布式锁

站长

2023年09月07日 22:48 · 阅读数 45

继上一篇 SpringBoot 整合 redis 踩坑日志之后，又学习了 redis 分布式锁，那为什么需要分布式锁？

redis 分布式锁原理

在传统单体应用单机部署的情况下，可以使用 Java 并发相关的锁，如 ReentrantLcok 或 synchronized 进行互斥控制。但是，随着业务发展的需要，原单体单机部署的系统，渐渐的被部署在多机器多JVM上同时提供服务，这使得原单机部署情况下的并发控制锁策略失效了，为了解决这个问题就需要一种跨JVM的互斥机制来控制共享资源的访问，这就是分布式锁要解决的问题。

分布式锁的实现条件

互斥性，和单体应用一样，要保证任意时刻，只能有一个客户端持有锁
可靠性，要保证系统的稳定性，不能产生死锁
一致性，要保证锁只能由加锁人解锁，不能产生A的加锁被B用户解锁的情况

分布式锁的实现

Redis 实现分布式锁不同的人可能有不同的实现逻辑，但是核心就是下面三个方法。

1.SETNXSETNX key val 当且仅当 key 不存在时，set 一个 key 为 val 的字符串，返回1；若 key存在，则什么都不做，返回0。

2.Expireexpire key timeout 为 key 设置一个超时时间，单位为second，超过这个时间锁会自动释放，避免死锁。

3.Deletedelete key 删除 key 。

原理图如下：

redis 分布式锁实战

项目代码结构图

导入依赖

在 pom.xml 中添加 starter-web、starter-aop、starter-data-redis 的依赖

<dependencies>
    <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>
</dependencies>

属性配置

在 application.properites 资源文件中添加 redis 相关的配置项

server:
  port: 1999
spring:
  datasource:
    url: jdbc:mysql://127.0.0.1:3306/mybatis-plus-test?useUnicode=true&characterEncoding=UTF-8&useSSL=false&serverTimezone=Asia/Shanghai
    driverClassName: com.mysql.cj.jdbc.Driver
    username: root
    password: root
  redis:
    host: 127.0.0.1
    port: 6379
    timeout: 5000ms
    password:
    database: 0
    jedis:
      pool:
        max-active: 50
        max-wait: 3000ms
        max-idle: 20
        min-idle: 2

注解

1、创建一个 CacheLock 注解，属性配置如下

prefix：缓存中 key 的前缀
expire：过期时间，此处默认为 5 秒
timeUnit：超时单位，此处默认为秒
delimiter： key 的分隔符，将不同参数值分割开来

    package com.tuhu.twosample.chen.distributed.annotation;
    
    import java.lang.annotation.*;
    import java.util.concurrent.TimeUnit;
    
    /**
     * 锁的注解
     * @author chendesheng
     * @create 2019/10/11 16:06
     */
    @Target({ElementType.METHOD})
    @Retention(RetentionPolicy.RUNTIME)
    @Documented
    @Inherited
    public @interface CacheLock {
    
        /**
         * redis 锁key的前缀
         *
         * @return redis 锁key的前缀
         */
        String prefix() default "";
    
        /**
         * 过期秒数,默认为5秒
         *
         * @return 轮询锁的时间
         */
        int expire() default 5;
    
        /**
         * 超时时间单位
         *
         * @return 秒
         */
        TimeUnit timeUnit() default TimeUnit.SECONDS;
    
        /**
         * <p>Key的分隔符（默认 :）</p>
         * <p>生成的Key：N:SO1008:500</p>
         *
         * @return String
         */
        String delimiter() default ":";
    
    }

2、 key 的生成规则是自己定义的，如果通过表达式语法自己得去写解析规则还是比较麻烦的，所以依旧是用注解的方式

package com.tuhu.twosample.chen.distributed.annotation;

import java.lang.annotation.*;

/**
 * 锁的参数
 * @author chendesheng
 * @create 2019/10/11 16:08
 */
@Target({ElementType.PARAMETER, ElementType.METHOD, ElementType.FIELD})
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
public @interface CacheParam {

    /**
     * 字段名称
     *
     * @return String
     */
    String name() default "";
}

key生成策略

1、接口

package com.tuhu.twosample.chen.distributed.common;

import org.aspectj.lang.ProceedingJoinPoint;

/**
 * key生成器
 * @author chendesheng
 * @create 2019/10/11 16:09
 */
public interface CacheKeyGenerator {

    /**
     * 获取AOP参数,生成指定缓存Key
     *
     * @param pjp PJP
     * @return 缓存KEY
     */
    String getLockKey(ProceedingJoinPoint pjp);
}

2、接口实现

package com.tuhu.twosample.chen.distributed.common;

import com.tuhu.twosample.chen.distributed.annotation.CacheLock;
import com.tuhu.twosample.chen.distributed.annotation.CacheParam;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;

import java.lang.annotation.Annotation;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.lang.reflect.Parameter;

/**
 * 通过接口注入的方式去写不同的生成规则
 * @author chendesheng
 * @create 2019/10/11 16:09
 */
public class LockKeyGenerator implements CacheKeyGenerator {

    @Override
    public 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 注解的属性,如果没有尝试着解析实体对象中的
        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())) {
            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();
    }


}

Lock拦截器（AOP）

熟悉 Redis 的朋友都知道它是线程安全的，我们利用它的特性可以很轻松的实现一个分布式锁，如opsForValue().setIfAbsent(key,value)它的作用就是如果缓存中没有当前 Key 则进行缓存同时返回 true 反之亦然；当缓存后给 key 在设置个过期时间，防止因为系统崩溃而导致锁迟迟不释放形成死锁；那么我们是不是可以这样认为当返回 true 我们认为它获取到锁了，在锁未释放的时候我们进行异常的抛出….

package com.tuhu.twosample.chen.distributed.interceptor;

import com.tuhu.twosample.chen.distributed.annotation.CacheLock;
import com.tuhu.twosample.chen.distributed.common.CacheKeyGenerator;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.util.StringUtils;

import java.lang.reflect.Method;

/**
 * @author chendesheng
 * @create 2019/10/11 16:11
 */
@Aspect
@Configuration
public class LockMethodInterceptor {

    @Autowired
    public LockMethodInterceptor(StringRedisTemplate lockRedisTemplate, CacheKeyGenerator cacheKeyGenerator) {
        this.lockRedisTemplate = lockRedisTemplate;
        this.cacheKeyGenerator = cacheKeyGenerator;
    }

    private final StringRedisTemplate lockRedisTemplate;
    private final CacheKeyGenerator cacheKeyGenerator;


    @Around("execution(public * *(..)) && @annotation(com.tuhu.twosample.chen.distributed.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 can't be null...");
        }
        final String lockKey = cacheKeyGenerator.getLockKey(pjp);
        try {
            //key不存在才能设置成功
            final Boolean success = lockRedisTemplate.opsForValue().setIfAbsent(lockKey, "");
            if (success) {
                lockRedisTemplate.expire(lockKey, lock.expire(), lock.timeUnit());
            } else {
                //按理来说 我们应该抛出一个自定义的 CacheLockException 异常;
                throw new RuntimeException("请勿重复请求");
            }
            try {
                return pjp.proceed();
            } catch (Throwable throwable) {
                throw new RuntimeException("系统异常");
            }
        } finally {

            //如果演示的话需要注释该代码;实际应该放开
            // lockRedisTemplate.delete(lockKey);

        }
    }
}

控制层

在接口方法上添加 @CacheLock(prefix = "test")，然后动态的值可以加上@CacheParam；生成后的新 key 将被缓存起来；（如：该接口 token = 1，那么最终的 key 值为 test:1，如果多个条件则依次类推）

package com.tuhu.twosample.chen.controller;

import com.tuhu.twosample.chen.distributed.annotation.CacheLock;
import com.tuhu.twosample.chen.distributed.annotation.CacheParam;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
/**
 * @author chendesheng
 * @create 2019/10/11 16:13
 */
@RestController
@RequestMapping("/chen/lock")
@Slf4j
public class LockController {
    @CacheLock(prefix = "test")
    @GetMapping("/test")
    public String query(@CacheParam(name = "token") @RequestParam String token) {
        return "success - " + token;
    }
}

主函数

需要注入前面定义好的 CacheKeyGenerator 接口具体实现 ….

package com.tuhu.twosample;

import com.tuhu.twosample.chen.distributed.common.CacheKeyGenerator;
import com.tuhu.twosample.chen.distributed.common.LockKeyGenerator;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
/**
 * @author chendesheng
 * @since 2019-08-06
 */
@SpringBootApplication
@MapperScan("com.baomidou.mybatisplus.samples.quickstart.mapper")
@MapperScan("com.tuhu.twosample.chen.mapper")
public class TwoSampleApplication {

  	public static void main(String[] args) {
        SpringApplication.run(TwoSampleApplication.class, args);
    }
    @Bean
    public CacheKeyGenerator cacheKeyGenerator() {
        return new LockKeyGenerator();
    }
}

测试

启动项目，在postman中输入url：<http://localhost:1999/chen/lock/test?token=1 >

第一次请求结果：

第二次请求结果：

等key过期了请求又恢复正常。

最后

但是这种分布式锁也存在着缺陷，如果A在setnx成功后，A成功获取锁了，也就是锁已经存到 Redis 里面了，此时服务器异常关闭或是重启，将不会执行我们的业务逻辑，也就不会设置锁的有效期，这样的话锁就不会释放了，就会产生死锁。所以还需要对锁进行优化，好好学习学习，嘎嘎嘎嘎。

转载自:https://juejin.cn/post/6844903961489588231