2022-09-28

字数统计: 7k字 | 阅读时长≈ 30分

Redis接口限流、分布式锁与幂等

一、概述

1、Redis概述

Redis参考文章：Redis6.0学习笔记

Redis 除了做缓存，还能干很多很多事情：分布式锁、限流、处理请求接口幂等性，本篇文章重点讲述SpringBoot通过注解和AOP的方式实现Redis的接口限流，Redis使用了Lua脚本实现原子操作；通过redis实现的分布式锁以及处理接口幂等等方案

2、功能介绍

2.1 Redis限流

限流就是限制API访问频率，当访问频率超过某个阈值时进行拒绝访问等操作

当然这是在代码层面进行的接口限流，现在分布式微服务接口限流基本是在网关处做接口限流/黑白名单等，例如Gateway/Nginx等，详情可以参考Nginx高级篇和SpringCloud Gateway 详解

2.2 分布式锁

为了保证一个方法在高并发情况下的同一时间只能被同一个线程执行，在传统单体应用单机部署的情况下，可以使用Java并发处理相关的API(如ReentrantLcok或synchronized)进行互斥控制。但是，随着业务发展的需要，原单体单机部署的系统被演化成分布式系统后，由于分布式系统多线程、多进程并且分布在不同机器上，这将使原单机部署情况下的并发控制锁策略失效，为了解决这个问题就需要一种跨JVM的互斥机制来控制共享资源的访问，这就是分布式锁要解决的问题

2.3 接口幂等

幂等性原本是数学上的概念，用在接口上就可以理解为：同一个接口，多次发出同一个请求，必须保证操作只执行一次。调用接口发生异常并且重复尝试时，总是会造成系统所无法承受的损失，所以必须阻止这种现象的发生

二、Redis接口限流实战

1、环境准备

首先我们创建一个 Spring Boot 工程，引入 Web 和 Redis 依赖，同时考虑到接口限流一般是通过注解来标记，而注解是通过 AOP 来解析的，所以我们还需要加上 AOP 的依赖，最终的依赖如下

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-data-redis</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>com.alibaba.nacos</groupId>
  <artifactId>nacos-client</artifactId>
</dependency>
<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-lang3</artifactId>
</dependency>

然后提前准备好一个** Redis 实例**，这里我们项目配置好之后，直接配置一下 Redis 的基本信息即可，如下：

1
2
3

spring.redis.host=localhost
spring.redis.port=6379
# spring.redis.password=123

2、限流注解

接下来我们创建一个限流注解，我们将限流分为两种情况：

针对当前接口的全局性限流，例如该接口可以在 1 分钟内访问 100 次
针对某一个 IP 地址的限流，例如某个 IP 地址可以在 1 分钟内访问 100 次

针对这两种情况，我们创建一个枚举类

public enum LimitType {
    /**
     * 默认策略全局限流
     */
    DEFAULT,
    /**
     * 根据请求者IP进行限流
     */
    IP
}

接下来我们来创建限流注解，第一个参数限流的 key，这个仅仅是一个前缀，将来完整的 key 是这个前缀再加上接口方法的完整路径，共同组成限流 key，这个 key 将被存入到 Redis 中

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface RateLimiter {
    /**
     * 限流key
     */
    String key() default "rate_limit:";

    /**
     * 限流时间,单位秒
     */
    int time() default 60;

    /**
     * 限流次数
     */
    int count() default 100;

    /**
     * 限流类型
     */
    LimitType limitType() default LimitType.DEFAULT;
}

将来哪个接口需要限流，就在哪个接口上添加 @RateLimiter 注解，然后配置相关参数即可

3、配置RedisTemplate

默认的 RedisTemplate 有一个小坑，就是序列化用的是 JdkSerializationRedisSerializer，直接用这个序列化工具将来存到 Redis 上的 key 和 value 都会莫名其妙多一些前缀，这就导致你用命令读取的时候可能会出错，此时当你在命令行操作的时候，get name 却获取不到你想要的数据，原因就是存到 redis 之后 name 前面多了一些字符，此时只能继续使用 RedisTemplate 将之读取出来

因为Redis限流用到了Lua脚本，因此需要改写我们自己的序列化方案，使用 Spring Boot 中默认的 jackson 序列化方式来解决

@Configuration
public class RedisConfig {
    // 编写自己的RedisTemplate
    @Bean
    @SuppressWarnings("all")
    public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
        RedisTemplate<Object, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(redisConnectionFactory);
        //序列化配置
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<>(Object.class);
        ObjectMapper om = new ObjectMapper();
        // 指定要序列化的域，field,get和set,以及修饰符范围，ANY是都有包括private和public
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        // 指定序列化输入的类型，类必须是非final修饰的，final修饰的类，比如String,Integer等会跑出异常
        // 序列化时会自动增加类类型，否则无法反序列化
        om.activateDefaultTyping(LaissezFaireSubTypeValidator.instance, ObjectMapper.DefaultTyping.NON_FINAL, JsonTypeInfo.As.PROPERTY);
        jackson2JsonRedisSerializer.setObjectMapper(om);
        //String的序列化
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
        // key采用String的序列化方式
        template.setKeySerializer(stringRedisSerializer);
        // hash采用String序列方式
        template.setHashKeySerializer(stringRedisSerializer);
        // value采用jackson
        template.setValueSerializer(jackson2JsonRedisSerializer);
        // hash的value采用jackson
        template.setHashValueSerializer(jackson2JsonRedisSerializer);
        template.afterPropertiesSet();
        return template;
    }
}

4、开发 Lua 脚本

Redis 中的一些原子操作我们可以借助 Lua 脚本来实现，想要调用 Lua 脚本，我们有两种不同的思路

在 Redis 服务端定义好 Lua 脚本，然后计算出来一个散列值，在 Java 代码中，通过这个散列值锁定要执行哪个 Lua 脚本
直接在 Java 代码中将 Lua 脚本定义好，然后发送到 Redis 服务端去执行

Spring Data Redis 中也提供了操作 Lua 脚本的接口，还是比较方便的，所以我们这里就采用第二种方案，我们在 resources 目录下新建 lua 文件夹专门用来存放 lua 脚本

local key = KEYS[1]
-- tonumber 把字符串转为数字
local count = tonumber(ARGV[1])
local time = tonumber(ARGV[2])
-- 执行具体的 redis 指令
local current = redis.call('get', key)
if current and tonumber(current) > count then
    return tonumber(current)
end
current = redis.call('incr', key)
if tonumber(current) == 1 then
    redis.call('expire', key, time)
end
return tonumber(current)

这个lua脚本执行流程

首先获取到传进来的 key 以及限流的 count 和时间 time
通过 get 获取到这个 key 对应的值，这个值就是当前时间窗内这个接口可以访问多少次
如果是第一次访问，此时拿到的结果为 nil，否则拿到的结果应该是一个数字，所以接下来就判断，如果拿到的结果是一个数字，并且这个数字还大于 count，那就说明已经超过流量限制了，那么直接返回查询的结果即可
如果拿到的结果为 nil，说明是第一次访问，此时就给当前 key 自增 1，然后设置一个过期时间
最后把自增 1 后的值返回

最后在Spring中加载这个Lua脚本

@Configuration
public class MyBean {

    @Bean
    public DefaultRedisScript<Long> limitScript() {
        DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
        redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("lua/limit.lua")));
        redisScript.setResultType(Long.class);
        return redisScript;
    }

}

5、全局类与工具类

由于过载的时候是抛异常出来，所以我们还需要一个全局异常处理，其他详细可以参考Spring Boot后端接口规范

@RestControllerAdvice
public class GlobalException {
    @ExceptionHandler(ServiceException.class)
    public Map<String,Object> serviceException(ServiceException e) {
        HashMap<String, Object> map = new HashMap<>();
        map.put("status", 500);
        map.put("message", e.getMessage());
        return map;
    }
}

IpUtils工具类，获取Ip或者Mac

public class IpUtils {

    /**
     * 获取当前网络ip
     *
     * @param request
     * @return
     */
    public static String getIpAddr(HttpServletRequest request) {
        String ipAddress = request.getHeader("X-Forwarded-For");
        if (ipAddress == null || ipAddress.length() == 0 || "unknown".equalsIgnoreCase(ipAddress)) {
            ipAddress = request.getHeader("X-Real-IP");
        }
        if (ipAddress == null || ipAddress.length() == 0 || "unknown".equalsIgnoreCase(ipAddress)) {
            ipAddress = request.getHeader("Proxy-Client-IP");
        }
        if (ipAddress == null || ipAddress.length() == 0 || "unknown".equalsIgnoreCase(ipAddress)) {
            ipAddress = request.getHeader("WL-Proxy-Client-IP");
        }
        if (ipAddress == null || ipAddress.length() == 0 || "unknown".equalsIgnoreCase(ipAddress)) {
            ipAddress = request.getRemoteAddr();
            if ("127.0.0.1".equals(ipAddress) || "0:0:0:0:0:0:0:1".equals(ipAddress)) {
                // 根据网卡取本机配置的IP
                InetAddress inet = null;
                try {
                    inet = InetAddress.getLocalHost();
                } catch (UnknownHostException e) {
                    e.printStackTrace();
                }
                ipAddress = inet.getHostAddress();
            }
        }
        // 对于通过多个代理的情况，第一个IP为客户端真实IP,多个IP按照','分割
        if (ipAddress != null && ipAddress.length() > 15) { // "***.***.***.***".length() = 15
            if (ipAddress.indexOf(",") > 0) {
                ipAddress = ipAddress.substring(0, ipAddress.indexOf(","));
            }
        }
        return ipAddress;
    }

    /**
     * 获得MAC地址
     *
     * @param ip
     * @return
     */
    public static String getMACAddress(String ip) {
        String str = "";
        String macAddress = "";
        try {
            Process p = Runtime.getRuntime().exec("nbtstat -A " + ip);
            InputStreamReader ir = new InputStreamReader(p.getInputStream());
            LineNumberReader input = new LineNumberReader(ir);
            for (int i = 1; i < 100; i++) {
                str = input.readLine();
                if (str != null) {
                    if (str.indexOf("MAC Address") > 1) {
                        macAddress = str.substring(str.indexOf("MAC Address") + 14, str.length());
                        break;
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace(System.out);
        }
        return macAddress;
    }

}

6、注解AOP解析

下面的切面就是拦截所有加了 @RateLimiter 注解的方法，在前置通知中对注解进行处理。

首先获取到注解中的 key、time 以及 count 三个参数
获取一个组合的 key，所谓的组合的 key，就是在注解的 key 属性基础上，再加上方法的完整路径，如果是 IP 模式的话，就再加上 IP 地址。以 IP 模式为例，最终生成的 key 类似这样：rate_limit:192.168.249.1-com.example.limiting.controller.HelloController-hello（如果不是 IP 模式，那么生成的 key 中就不包含 IP 地址）
将生成的 key 放到集合中
通过 redisTemplate.execute 方法取执行一个 Lua 脚本，第一个参数是脚本所封装的对象，第二个参数是 key，对应了脚本中的 KEYS，后面是可变长度的参数，对应了脚本中的 ARGV
将 Lua 脚本执行的结果与 count 进行比较，如果大于 count，就说明过载了，抛异常就行了

@Aspect
@Component
public class RateLimiterAspect {
    private static final Logger log = LoggerFactory.getLogger(RateLimiterAspect.class);

    @Autowired
    private RedisTemplate<Object, Object> redisTemplate;

    @Autowired
    private RedisScript<Long> limitScript;

    @Before("@annotation(rateLimiter)")
    public void doBefore(JoinPoint point, RateLimiter rateLimiter) throws Throwable {
        // 获取注解参数
        String key = rateLimiter.key();
        int time = rateLimiter.time();
        int count = rateLimiter.count();

        // 获取redis的key
        String combineKey = getCombineKey(rateLimiter, point);
        List<Object> keys = Collections.singletonList(combineKey);
        try {
            Long number = redisTemplate.execute(limitScript, keys, count, time);
            if (number==null || number.intValue() > count) {
                throw new ServiceException("访问过于频繁，请稍候再试");
            }
            log.info("限制请求'{}',当前请求'{}',缓存key'{}'", count, number.intValue(), combineKey);
        } catch (ServiceException e) {
            throw e;
        } catch (Exception e) {
            throw new RuntimeException("服务器限流异常，请稍候再试");
        }
    }

    public String getCombineKey(RateLimiter rateLimiter, JoinPoint point) {
        StringBuffer stringBuffer = new StringBuffer(rateLimiter.key());
        if (rateLimiter.limitType() == LimitType.IP) {
            // 这个方法可以获取到当前线程的request和response
            stringBuffer.append(IpUtils.getIpAddr(((ServletRequestAttributes) RequestContextHolder.currentRequestAttributes()).getRequest())).append("-");
        }
        MethodSignature signature = (MethodSignature) point.getSignature();
        Method method = signature.getMethod();
        Class<?> targetClass = method.getDeclaringClass();
        stringBuffer.append(targetClass.getName()).append("-").append(method.getName());
        return stringBuffer.toString();
    }
}

7、接口测试

进行简单的测试，下面每一个 IP 地址，在 5 秒内只能访问 3 次

@RestController
public class HelloController {
    @GetMapping("/hello")
    @RateLimiter(time = 5,count = 3,limitType = LimitType.IP)
    public String hello() {
        return "hello>>>"+new Date();
    }
}

三、Redis分布式锁

1、简介

分布式锁其实就是，控制分布式系统不同进程共同访问共享资源的一种锁的实现。如果不同的系统或同一个系统的不同主机之间共享了某个临界资源，往往需要互斥来防止彼此干扰，以保证一致性。

分布式锁一般都使用Redis来实现，大概有以下几种方案，可以参考redis分布式锁

SETNX + EXPIRE
SETNX + value值是（系统时间+过期时间）
使用Lua脚本(包含SETNX + EXPIRE两条指令)
SET的扩展命令（SET EX PX NX）
SET EX PX NX + 校验唯一随机值，再释放锁 (推荐)
开源框架Redisson (推荐)
多机实现的分布式锁Redlock (推荐)

2、AOP分布式锁原理

2.1 实现流程

新建注解 @interface，在注解里设定入参标志
增加 AOP 切点，扫描特定注解
建立 @Aspect 切面任务，注册 bean 和拦截特定方法
特定方法参数 ProceedingJoinPoint，对方法 pjp.proceed() 前后进行拦截
切点前进行加锁，任务执行后进行删除 key

在这里插入图片描述

2.2 核心步骤

加锁

使用了 StringRedisTemplate 的 opsForValue.setIfAbsent 方法，判断是否有 key，设定一个随机数 UUID.random().toString，生成一个随机数作为 value。从 redis 中获取锁之后，对 key 设定 expire 失效时间，到期后自动释放锁。按照这种设计，只有第一个成功设定 Key 的请求，才能进行后续的数据操作，后续其它请求由于无法获得🔐资源，将会失败结束。

超时问题

担心 pjp.proceed() 切点执行的方法太耗时，导致 Redis 中的 key 由于超时提前释放了。例如，线程 A 先获取锁，proceed 方法耗时，超过了锁超时时间，到期释放了锁，这时另一个线程 B 成功获取 Redis 锁，两个线程同时对同一批数据进行操作，导致数据不准确。

锁续时操作(任务不完成，锁不释放)

维护了一个定时线程池 ScheduledExecutorService，每隔 2s 去扫描加入队列中的 Task，判断是否失效时间是否快到了，公式为：【失效时间】<= 【当前时间】+【失效间隔（三分之一超时）】

3、AOP分布式锁实战

3.1 业务属性枚举设定

环境与Redis限流一样，首先创建注解

public enum RedisLockTypeEnum {
    /**
     * 自定义 key 前缀
     */
    ONE("Business1", "Test1"),

    TWO("Business2", "Test2");
    private final String code;
    private final String desc;
    RedisLockTypeEnum(String code, String desc) {
        this.code = code;
        this.desc = desc;
    }
    public String getCode() {
        return code;
    }
    public String getDesc() {
        return desc;
    }
    public String getUniqueKey(String key) {
        return String.format("%s:%s", this.getCode(), key);
    }
}

3.2 任务队列保存参数

@Data
public class RedisLockDefinitionHolder {
    /**
     * 业务唯一 key
     */
    private String businessKey;
    /**
     * 加锁时间 (秒 s)
     */
    private Long lockTime;
    /**
     * 上次更新时间（ms）
     */
    private Long lastModifyTime;
    /**
     * 保存当前线程
     */
    private Thread currentTread;
    /**
     * 总共尝试次数
     */
    private int tryCount;
    /**
     * 当前尝试次数
     */
    private int currentCount;
    /**
     * 更新的时间周期（毫秒）,公式 = 加锁时间（转成毫秒） / 3
     */
    private Long modifyPeriod;
    public RedisLockDefinitionHolder(String businessKey, Long lockTime, Long lastModifyTime, Thread currentTread, int tryCount) {
        this.businessKey = businessKey;
        this.lockTime = lockTime;
        this.lastModifyTime = lastModifyTime;
        this.currentTread = currentTread;
        this.tryCount = tryCount;
        this.modifyPeriod = lockTime * 1000 / 3;
    }
}

3.3 拦截的注解名声明

@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.METHOD, ElementType.TYPE})
public @interface RedisLockAnnotation {
    /**
     * 特定参数识别，默认取第 0 个下标
     */
    int lockFiled() default 0;
    /**
     * 超时重试次数
     */
    int tryCount() default 3;
    /**
     * 自定义加锁类型
     */
    RedisLockTypeEnum typeEnum();
    /**
     * 释放时间，秒 s 单位
     */
    long lockTime() default 30;
}

3.4 核心切面拦截

解析注解参数，获取注解值和方法上的参数值
redis 加锁并且设置超时时间
将本次 Task 信息加入「延时」队列中，进行续时，方式提前释放锁
加了一个线程中断标志
结束请求，finally 中释放锁

@Component
@Aspect
@Slf4j
public class RedisLockAspect {

    @Autowired
    StringRedisTemplate redisTemplate;

    /**
     * @annotation 中的路径表示拦截特定注解
     */
    @Pointcut("@annotation(com.example.redislock.anno.RedisLockAnnotation)")
    public void redisLockPC() {
    }


    @Around(value = "redisLockPC()")
    public Object around(ProceedingJoinPoint pjp) throws Throwable {
        Method method = null;
        // 解析参数
        //判断注解是否method 上
        if (pjp.getSignature() instanceof MethodSignature) {
            MethodSignature signature = (MethodSignature) pjp.getSignature();
            method = signature.getMethod();
        } else {
            return null;
        }
        RedisLockAnnotation annotation = method.getAnnotation(RedisLockAnnotation.class);
        RedisLockTypeEnum typeEnum = annotation.typeEnum();
        Object[] params = pjp.getArgs();
        String ukString = params[annotation.lockFiled()].toString();
        // 省略很多参数校验和判空
        String businessKey = typeEnum.getUniqueKey(ukString);
        String uniqueValue = UUID.randomUUID().toString();
        // 加锁
        Object result = null;
        try {
            boolean isSuccess = Boolean.TRUE.equals(redisTemplate.opsForValue().setIfAbsent(businessKey, uniqueValue));
            if (!isSuccess) {
                throw new Exception("You can't do it，because another has get the lock =-=");
            }
            log.info("get the lock, businessKey is [" + businessKey + "]");
            redisTemplate.expire(businessKey, annotation.lockTime(), TimeUnit.SECONDS);
            Thread currentThread = Thread.currentThread();
            // 将本次 Task 信息加入「延时」队列中
            ScheduledExecutorService.holderList.add(new RedisLockDefinitionHolder(businessKey, annotation.lockTime(), System.currentTimeMillis(),
                    currentThread, annotation.tryCount()));
            // 执行业务操作
            result = pjp.proceed();
            // 线程被中断，抛出异常，中断此次请求
            if (currentThread.isInterrupted()) {
                throw new InterruptedException("You had been interrupted =-=");
            }
        } catch (InterruptedException e ) {
            log.error("Interrupt exception, rollback transaction", e);
            throw new Exception("Interrupt exception, please send request again");
        } catch (Exception e) {
            log.error("has some error, please check again", e);
        } finally {
            // 请求结束后，强制删掉 key，释放锁
            redisTemplate.delete(businessKey);
            log.info("release the lock, businessKey is [" + businessKey + "]");
        }
        return result;
    }
    
}

3.5 续时操作

这里加了「线程中断」Thread#interrupt，希望超过重试次数后，能让线程中断（仅供参考）

@Slf4j
@Service
public class ScheduledExecutorService {

    @Autowired
    StringRedisTemplate redisTemplate;


    // 扫描的任务队列
    public static ConcurrentLinkedQueue<RedisLockDefinitionHolder> holderList = new ConcurrentLinkedQueue();
    /**
     * 线程池，维护keyAliveTime
     */
    private static final ScheduledThreadPoolExecutor SCHEDULER = new ScheduledThreadPoolExecutor(1,
            new BasicThreadFactory.Builder().namingPattern("redisLock-schedule-pool").daemon(true).build());
    {
        // 两秒执行一次「续时」操作
        SCHEDULER.scheduleAtFixedRate(() -> {
            // 这里记得加 try-catch，否者报错后定时任务将不会再执行=-=
            Iterator<RedisLockDefinitionHolder> iterator = holderList.iterator();
            while (iterator.hasNext()) {
                RedisLockDefinitionHolder holder = iterator.next();
                // 判空
                if (holder == null) {
                    iterator.remove();
                    continue;
                }
                // 判断 key 是否还有效，无效的话进行移除
                if (redisTemplate.opsForValue().get(holder.getBusinessKey()) == null) {
                    iterator.remove();
                    continue;
                }
                // 超时重试次数，超过时给线程设定中断
                if (holder.getCurrentCount() > holder.getTryCount()) {
                    holder.getCurrentTread().interrupt();
                    iterator.remove();
                    continue;
                }
                // 判断是否进入最后三分之一时间
                long curTime = System.currentTimeMillis();
                boolean shouldExtend = (holder.getLastModifyTime() + holder.getModifyPeriod()) <= curTime;
                if (shouldExtend) {
                    log.info("超时增加");
                    holder.setLastModifyTime(curTime);
                    redisTemplate.expire(holder.getBusinessKey(), holder.getLockTime(), TimeUnit.SECONDS);
                    log.info("businessKey : [" + holder.getBusinessKey() + "], try count : " + holder.getCurrentCount());
                    holder.setCurrentCount(holder.getCurrentCount() + 1);
                }
            }
        }, 0, 2, TimeUnit.SECONDS);
    }

}

3.6 测试

在一个入口方法中，使用该注解，然后在业务中模拟耗时请求，使用了 Thread#sleep。使用时，在方法上添加该注解，然后设定相应参数即可，根据 typeEnum 可以区分多种业务，限制该业务被同时操作

@RestController
@Slf4j
public class TestController {

    @GetMapping("/")
    @RedisLockAnnotation(typeEnum = RedisLockTypeEnum.ONE, lockTime = 3)
    public String testRedisLock(@RequestParam("userId") Long userId) {
        try {
            log.info("睡眠执行前");
            Thread.sleep(20000);
            log.info("睡眠执行后");
        } catch (Exception e) {
            // log error
            log.info("has some error", e);
        }
        return null;
    }
}

4、Redission分布式锁(AOP实现)

Redission地址：https://github.com/redisson/redisson

首先需要引入相关依赖，这里需要额外引入redission依赖

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.17.6</version>
</dependency>

在applicatiion.properties创建参数

spring.redis.host=localhost
spring.redis.port=6379
# spring.redis.password=123
spring.redis.database=0
spring.redis.timeout=5000ms
#redisson客户端连接超时时间(ms)
redisson.timeout=10000

4.1 注解创建

@Retention(value = RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
public @interface RedisDistributedLock {

    /**
     * 要锁哪几个位置的参数，默认不锁参数
     * (如果锁参数, 需要指定参数的索引比如锁第一个参数和第二个参数则传{0, 1}
     * 锁参数之后, 锁的key就会拼接此参数
     */
    int[] lockIndex() default {-1};

    /**
     * 默认包名加方法名
     *
     * @return
     */
    String key() default "";

    /**
     * 过期时间 单位：毫秒
     * <pre>
     *     过期时间一定是要长于业务的执行时间.
     * </pre>
     */
    long expire() default 30000;

    /**
     * 获取锁超时时间 单位：毫秒
     * <pre>
     *     结合业务,建议该时间不宜设置过长,特别在并发高的情况下.
     * </pre>
     */
    long timeout() default 3000;


    /**
     * 时间类型，默认毫秒
     */
    TimeUnit timeUnit() default TimeUnit.MILLISECONDS;

}

4.2 创建切面增强

@Component
@Aspect
@Slf4j
public class RedisLockAop {

    @Resource
    private RedissonClient redissonClient;

    @Pointcut("@annotation(com.example.redislock.anno.RedisDistributedLock)")
    public void myAdvice() {
    }

    @Around("myAdvice()")
    public Object around(ProceedingJoinPoint proceedingJoinPoint) throws Throwable {
        // 获取注解
        MethodSignature signature = (MethodSignature) proceedingJoinPoint.getSignature();
        RedisDistributedLock annotation = signature.getMethod().getAnnotation(RedisDistributedLock.class);

        // 生成key
        StringBuilder keyBuilder = new StringBuilder(getKey(proceedingJoinPoint, annotation));

        // 如果锁参数, 需要将参数拼接到key上
        Object[] args = proceedingJoinPoint.getArgs();
        int[] lockIndex = annotation.lockIndex();
        if (lockIndex.length > 0 && lockIndex[0] >= 0) {
            for (int index : lockIndex) {
                if (index >= args.length || index < 0) {
                    throw new RuntimeException("参数索引lockIndex: " + index + " 异常");
                }
                keyBuilder.append(".").append(args[index].toString());
            }
        }

        // 防止key值太长，用根据其生成的hash值做key
        // String lockKey = DigestUtils.md5DigestAsHex(keyBuilder.toString().getBytes());
        String key = keyBuilder.toString();


        Boolean success = null;
        RLock lock = redissonClient.getLock(key);
        try {
            //lock提供带timeout参数，timeout结束强制解锁，防止死锁
            success = lock.tryLock(annotation.timeout(), annotation.expire(), annotation.timeUnit());
            if (success) {
                log.info(Thread.currentThread().getName() + " 加锁成功");
                // 放行方法执行
                return proceedingJoinPoint.proceed();
            }
            log.info(Thread.currentThread().getName() + " 加锁失败");
            throw new RuntimeException("操作频繁, 稍后重试"); // 此处可以用return 返回错误 需要跟切的方法的返回值保持一致
        } catch (Exception e) {
            throw e;
        } finally {
            if (Boolean.TRUE.equals(success)) {
                lock.unlock();
                log.info(Thread.currentThread().getName() + " 解锁成功");
            }
        }
    }

    private String getKey(ProceedingJoinPoint proceedingJoinPoint, RedisDistributedLock annotation) {
        if (!StringUtils.isEmpty(annotation.key())) {
            return annotation.key();
        }
        return proceedingJoinPoint.getSignature().getDeclaringTypeName() + "." + proceedingJoinPoint.getSignature()
                .getName();
    }
}

4.3 测试

@RestController
@Slf4j
public class TestController {

    
    /**
     * 手动加锁模拟 设置锁的时间为300秒 5分钟, 在5分钟之内若方法没有执行完成则自动解锁, 获取锁的等待时间为2秒
     */
    @GetMapping("/test")
    @RedisDistributedLock(timeUnit = TimeUnit.SECONDS, expire = 300, timeout = 2)
    public String test(String id) {
        System.out.println("id = [" + id + "]");
        try {
            Thread.sleep(10 * 1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "0";
    }
}

4.4 Redission其他

Redission还有以下几种锁以及集群操作，详情可以参考：springboot整合redission分布式锁的实现方式含集群解决方案（技术篇）

可重入锁（Reentrant Lock）
公平锁（Fair Lock）
读写锁（ReadWriteLock）
信号量（Semaphore）
闭锁（CountDownLatch）

四、Redis接口幂等

1、介绍

幂等性，就是只多次操作的结果是一致的

产生的问题

前端重复提交。比如这个业务处理需要2秒钟，我在2秒之内，提交按钮连续点了3次，如果非幂等性接口，那么后端就会处理3次。如果是查询，自然是没有影响的，因为查询本身就是幂等操作，但如果是新增，本来只是新增1条记录的，连点3次，就增加了3条，这显然不行。
响应超时而导致请求重试：在微服务相互调用的过程中，假如订单服务调用支付服务，支付服务支付成功了，但是订单服务接收支付服务返回的信息时超时了，于是订单服务进行重试，又去请求支付服务，结果支付服务又扣了一遍用户的钱。

解决方案

数据库记录状态机制：即每次操作前先查询状态，根据数据库记录的状态来判断是否要继续执行操作。比如订单服务调用支付服务，每次调用之前，先查询该笔订单的支付状态，从而避免重复操作。
token机制：请求业务接口之前，先请求token接口（会将生成的token放入redis中）获取一个token，然后请求业务接口时，带上token。在进行业务操作之前，我们先获取请求中携带的token，看看在redis中是否有该token，有的话，就删除，删除成功说明token校验通过，并且继续执行业务操作；如果redis中没有该token，说明已经被删除了，也就是已经执行过业务操作了，就不让其再进行业务操作。大致流程如下：

在这里插入图片描述

2、防重 Token 令牌流程

服务端提供获取 Token 的接口，该 Token 可以是一个序列号，也可以是一个分布式 ID 或者 UUID 串
客户端调用接口获取 Token，这时候服务端会生成一个 Token 串
然后将该串存入 Redis 数据库中，以该 Token 作为 Redis 的键（注意设置过期时间）
将 Token 返回到客户端，客户端拿到后应存到表单隐藏域中
客户端在执行提交表单时，把 Token 存入到 Headers 中，执行业务请求带上该 Headers
服务端接收到请求后从 Headers 中拿到 Token，然后根据 Token 到 Redis 中查找该 key 是否存在
服务端根据 Redis 中是否存该 key 进行判断，如果存在就将该 key 删除，然后正常执行业务逻辑。如果不存在就抛异常，返回重复提交的错误信息

3、放重Token实战

3.1 注解创建

@Target(ElementType.METHOD)  //使用于方法
@Retention(RetentionPolicy.RUNTIME) //运行时
public @interface ApiIdempotent {
}

3.2 配置返回渲染

public class ServletUtils
{
    /**
     * 将字符串渲染到客户端
     *
     * @param response 渲染对象
     * @param string 待渲染的字符串
     * @return null
     */
    public static String renderString(HttpServletResponse response, String string) {
        try
        {
            response.setStatus(200);
            response.setContentType("application/json");
            response.setCharacterEncoding("utf-8");
            response.getWriter().print(string);
        }
        catch (IOException e)
        {
            e.printStackTrace();
        }
        return null;
    }

}

3.3 放重Token生成与验证

@Slf4j
@Service
public class TokenUtilService {

    @Autowired
    private StringRedisTemplate redisTemplate;

    /**
     * 存入 Redis 的 Token 键的前缀
     */
    private static final String IDEMPOTENT_TOKEN_PREFIX = "idempotent_token:";

    /**
     * 创建 Token 存入 Redis，并返回该 Token
     *
     * @param value 用于辅助验证的 value 值
     * @return 生成的 Token 串
     */
    public String generateToken(String value) {
        // 实例化生成 ID 工具对象
        String token = UUID.randomUUID().toString();
        // 设置存入 Redis 的 Key
        String key = IDEMPOTENT_TOKEN_PREFIX + token;
        // 存储 Token 到 Redis，且设置过期时间为5分钟
        redisTemplate.opsForValue().set(key, value, 5, TimeUnit.MINUTES);
        // 返回 Token
        return token;
    }

    /**
     * 验证 Token 正确性
     *
     * @param token token 字符串
     * @param value value 存储在Redis中的辅助验证信息
     * @return 验证结果
     */
    public boolean validToken(String token, String value) {
        // 设置 Lua 脚本，其中 KEYS[1] 是 key，KEYS[2] 是 value
        String script = "if redis.call('get', KEYS[1]) == KEYS[2] then return redis.call('del', KEYS[1]) else return 0 end";
        RedisScript<Long> redisScript = new DefaultRedisScript<>(script, Long.class);
        // 根据 Key 前缀拼接 Key
        String key = IDEMPOTENT_TOKEN_PREFIX + token;
        // 执行 Lua 脚本
        Long result = redisTemplate.execute(redisScript, Arrays.asList(key, value));
        // 根据返回结果判断是否成功成功匹配并删除 Redis 键值对，若果结果不为空和0，则验证通过
        if (result != null && result != 0L) {
            log.info("验证 token={},key={},value={} 成功", token, key, value);
            return true;
        }
        log.info("验证 token={},key={},value={} 失败", token, key, value);
        return false;
    }

}

3.4、配置拦截器

首先创建自定义拦截器

@Component
public class ApiIdempotentInterceptor implements HandlerInterceptor {

    @Autowired
    private TokenUtilService tokenUtilService;

    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
        //校验是否有执行方法
        if (!(handler instanceof HandlerMethod)) {
            return true;//若没有对应的方法执行器，就直接放行
        }
        HandlerMethod handlerMethod = (HandlerMethod) handler;
        Method method = handlerMethod.getMethod();
        ApiIdempotent annotation = method.getAnnotation(ApiIdempotent.class);
        //若是没有幂等性注解直接放行
        if (annotation != null) {
            //解析对应的请求头
            String token = request.getHeader("token");
            if (ObjectUtils.isEmpty(token)) {
                ServletUtils.renderString(response, "请携带token令牌");
                return false;
            }
            //若是校验失败直接进行响应
            if (!tokenUtilService.validToken(token, "shawn")) {
                ServletUtils.renderString(response, "重复提交失败！");
                return false;
            }
        }
        return true;
    }
}

配置spring拦截器

@Configuration
public class WebConfiguration extends WebMvcConfigurationSupport {

    @Autowired
    private ApiIdempotentInterceptor apiIdempotentInterceptor;

    @Override
    public void addInterceptors(InterceptorRegistry registry) {
        registry.addInterceptor(apiIdempotentInterceptor);
        super.addInterceptors(registry);
    }

    //    http请求时编码
    @Bean
    public HttpMessageConverter<String> responseBodyConverter() {
        StringHttpMessageConverter converter = new StringHttpMessageConverter(
                StandardCharsets.UTF_8);
        return converter;
    }

    /**
     * 系统配置参数编码
     * @param converters
     */
    @Override
    protected void configureMessageConverters(List<HttpMessageConverter<?>> converters) {
        super.configureMessageConverters(converters);
        converters.add(responseBodyConverter());
    }

}

3.5 测试

@RestController
@Slf4j
public class TestController {

   
    /**
     * 获取 Token 接口
     *
     * @return Token 串
     */
    @GetMapping("/token")
    public String getToken() {
        // 获取用户信息（这里使用模拟数据）
        // 注：这里存储该内容只是举例，其作用为辅助验证，使其验证逻辑更安全，如这里存储用户信息，其目的为:
        // - 1)、使用"token"验证 Redis 中是否存在对应的 Key
        // - 2)、使用"用户信息"验证 Redis 的 Value 是否匹配。
        String userInfo = "changlu";
        // 获取 Token 字符串，并返回
        return tokenService.generateToken(userInfo);
    }

    /**
     * 接口幂等性测试接口
     *
     * @param token 幂等 Token 串
     * @return 执行结果
     */
    @GetMapping("/testToken")
    public Object test1(@RequestHeader(value = "token") String token) {
        // 获取用户信息（这里使用模拟数据）
        String userInfo = "shawn";
        // 根据 Token 和与用户相关的信息到 Redis 验证是否存在对应的信息
        boolean result = tokenService.validToken(token, userInfo);
        // 根据验证结果响应不同信息
        return result ? "正常调用" : "重复调用";
    }

    /**
     * 接口式的放重
     */
    @ApiIdempotent
    @GetMapping("/testToken2")
    public Object test2() {
        return "sucess";
    }

}

参考文章：

SpringBoot + 一个注解，轻松实现 Redis 分布式锁

springboot整合redission分布式锁的实现方式含集群解决方案（技术篇）

-------------本文结束^-^感谢您的阅读-------------

本文作者： Shawn
发布时间： 2022-09-28
最后更新： 2023-03-17
本文标题： Redis接口限流、分布式锁与幂等
本文链接： https://blog.shawncoding.top/posts/dcc28633.html
版权声明： 本作品采用 CC BY-NC-SA 4.0 许可协议进行许可。转载请注明出处！