背景

业务系统中的防重复提交都是由前端控制，后端在某些地方做了相应的业务逻辑相关的判断，但当某些情况下，前后端的判断都会失效，所以这里引入后端的接口防重复提交校验。

方案

由于需要限制的是部分接口，因此使用AOP+注解+Redis的方式来实现。AOP+注解的方式更加灵活，在需要限制的接口上加上注解即可。Redis则可以使防重复提交在分布式系统中使用。由于业务的特殊性，需要实现：1.同一个用户不能重复访问同一个接口；2.不同的用户不能以相同的参数同时访问同一个接口

实现

1.定义注解

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface RepeatSubmitVerify {
    /**
     * 设置请求锁定时间，单位：秒
     *
     * @return
     */
    int lockTime() default 10;

    /**
     * 参数名称
     *
     * @return
     */
    String paramString();
}

2.定义切面

@Aspect
@Component
@Slf4j
public class RepeatSubmitAspect {

    @Autowired
    private RedisLock redisLock;

    @Pointcut("@annotation(repeatSubmitVerify)")
    public void pointCut(RepeatSubmitVerify repeatSubmitVerify) {
    }

    @Around("pointCut(repeatSubmitVerify)")
    public Object around(ProceedingJoinPoint pjp, RepeatSubmitVerify repeatSubmitVerify) throws Throwable {
        int lockSeconds = repeatSubmitVerify.lockTime();

        String param = "{}";
        if (StringUtils.isNotBlank(repeatSubmitVerify.paramString())) {
            param = getKey(repeatSubmitVerify.paramString(), pjp);
        }

        JSONObject paramJson = JSON.parseObject(param);
        String token = paramJson.getString("token");

        // 这里需要设置两个lockKey，分别实现两个目的
        String key_prefix = pjp.getSignature().getDeclaringType().getSimpleName() + "_";
        // 接口名称+请求参数，避免不同的用户以相同的参数访问同时访问同一个接口
        String key1 = key_prefix + DigestUtils.md5Hex(paramJson.toJSONString());
        // 接口名称+用户token，避免同一个用户重复访问同一个接口
        String key2 = key_prefix + DigestUtils.md5Hex(token);
        // 上锁与解锁应该由同一个线程来执行，而不能其他线程来执行解锁，否则可能会出现错误解锁
        String clientId = getClientId();

		// 上锁
        Boolean isSuccess = redisLock.tryLock(key1, key2, clientId, lockSeconds);
        Assert.notNull(isSuccess, "系统访问校验异常");
        if (isSuccess) {
            log.info("tryLock success, key1 = [{}], key2 = [{}], clientId = [{}]", key1, key2, clientId);
            // 获取锁成功
            Object result;

            try {
                // 执行进程
                result = pjp.proceed();
            } finally {
                // 解锁
                Boolean releaseLock = redisLock.releaseLock(key1, key2, clientId);
                log.info("releaseLock success, key1 = [{}], key2 = [{}], clientId = [{}], result = [{}]", key1, key2, clientId, releaseLock);
            }

            return result;

        } else {
            // 获取锁失败，认为是重复提交的请求
            
            // 解锁这一步可以省略，在最终方案确定之前，上两把锁是分两步进行的，
            // 这样就会在重复请求时，其中一把上锁成功，而另一把不成功，判定上锁是失败的，
            // 因此要将成功的一把锁进行解锁，需要执行以下步骤
            // 而最终方案是通过lua脚本执行，要么成功上锁，两把锁都成功，要么失败，两把锁都失败，所以这一步可以省略
            Boolean releaseLock = redisLock.releaseLock(key1, key2, clientId);
            log.info("releaseLock success, key1 = [{}], key2 = [{}], clientId = [{}], result = [{}]", key1, key2, clientId, releaseLock);
            
            return errorData("操作已受理，请勿重复操作！");
        }

    }
	//获取请求参数
    private String getKey(String key, JoinPoint joinPoint) {
        Method method = ((MethodSignature) (joinPoint.getSignature())).getMethod();
        String[] paramenterNames = new LocalVariableTableParameterNameDiscoverer().getParameterNames(method);
        ExpressionParser parser = new SpelExpressionParser();
        Expression expression = parser.parseExpression(key);
        EvaluationContext context = new StandardEvaluationContext();
        Object[] args = joinPoint.getArgs();
        if (args.length <= 0) {
            return "";
        }
        for (int i = 0; i < args.length; i++) {
            context.setVariable(paramenterNames[i], args[i]);
        }
        return expression.getValue(context, String.class);
    }
	//获取每一次访问的UUID
    private String getClientId() {
        return UUID.randomUUID().toString();
    }
}

@Service
@Slf4j
public class RedisLock {
	//lua脚本中的返回值判断需要特别注意
    private static final String RELEASE_LOCK_SCRIPT = "local p1 local p2 if redis.call('get', KEYS[1]) == KEYS[3] then p1 = redis.call('del', KEYS[1]) else p1 = 1 end if redis.call('get', KEYS[2]) == KEYS[3] then p2 = redis.call('del', KEYS[2]) else p2 = 1 end return p1~=0 or p2~=0";
    private static final String TRY_LOCK_SCRIPT = "if redis.call('exists', KEYS[1]) == 1 or redis.call('exists', KEYS[2]) == 1 then return 0 else redis.call('setex', KEYS[1], KEYS[4], KEYS[3]) redis.call('setex', KEYS[2], KEYS[4], KEYS[3]) return 1 end";
  
    @Autowired
    private StringRedisTemplate redisTemplate;

    /**
     * 该加锁方法仅针对单实例 Redis 可实现分布式加锁
     * 对于 Redis 集群则无法使用
     * <p>
     * 支持重复，线程安全
     *
     * @param lockKey1 加锁键
     * @param lockKey2 加锁键
     * @param clientId 加锁客户端唯一标识(采用UUID)
     * @param seconds  锁过期时间
     * @return
     */
    public Boolean tryLock(String lockKey1, String lockKey2, String clientId, long seconds) {
        try {
            Boolean result = redisTemplate.execute(new DefaultRedisScript<>(TRY_LOCK_SCRIPT, Boolean.class), Arrays.asList(lockKey1, lockKey2, clientId, String.valueOf(seconds)));
            log.info("tryLock, lockKey1 = [{}], lockKey2 = [{}], result = [{}], lockVal1 = [{}], lockVal2 = [{}]", lockKey1, lockKey2, result, redisTemplate.opsForValue().get(lockKey1), redisTemplate.opsForValue().get(lockKey1));
            return result;
            // return redisTemplate.opsForValue().setIfAbsent(lockKey1, clientId, seconds, TimeUnit.SECONDS)
            //         && redisTemplate.opsForValue().setIfAbsent(lockKey2, clientId, seconds, TimeUnit.SECONDS);
        } catch (Exception e) {
            return false;
        }
    }

    /**
     * 与 tryLock 相对应，用作释放锁
     *
     * @param lockKey1
     * @param lockKey2
     * @param clientId
     * @return
     */
    public Boolean releaseLock(String lockKey1, String lockKey2, String clientId) {
        try {
            return redisTemplate.execute(new DefaultRedisScript<>(RELEASE_LOCK_SCRIPT, Boolean.class), Arrays.asList(lockKey1, lockKey2, clientId));
        } catch (Exception e) {
            log.error("{}", e);
            return false;
        }
    }

3.在需要处理的接口上打注解

参考资料

Redis eval命令踩得那些坑：https://github.com/nethibernate/blog/issues/7
Redis分布式锁的正确实现方式：https://www.cnblogs.com/linjiqin/p/8003838.html
CentOS7 安装lua环境：https://blog.csdn.net/houjixin/article/details/46634847
参考Github代码仓库：https://github.com/MissDistin/repeat-submit-intercept

接口重复提交：是由于网络等原因，造成一瞬间发送多个请求，造成insert，update操作，多次数据被修改。如果多个请求时间间隔足够的小，那么可以理解为并发问题；即并发情况下，只有一次操作成功。

实现原理： 自定义注解+AOP切面+分布式锁。
1.自定义注解：

import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

/**
 * 防止重复提交的注解
 *
 * @author yxh
 */
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.METHOD})
public @interface AvoidRepeatSubmit {

    long lockTime() default 1000;

}

2.分布式锁：

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.JedisCommands;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;

/**
 * Created by yxh on 2022/1/10 18:02
 */
@Component
public class RedisDistributedLock {

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    public static final String UNLOCK_LUA;

    static {
        StringBuilder sb = new StringBuilder();
        sb.append("if redis.call(\"get\",KEYS[1]) == ARGV[1] ");
        sb.append("then ");
        sb.append("    return redis.call(\"del\",KEYS[1]) ");
        sb.append("else ");
        sb.append("    return 0 ");
        sb.append("end ");
        UNLOCK_LUA = sb.toString();
    }

    private final Logger logger = LoggerFactory.getLogger(RedisDistributedLock.class);

    public boolean setLock(String key, String clientId, long expire) {
        try {
            //该的值只能取 EX 或者 PX，代表数据过期时间的单位，EX 代表秒，PX 代表毫秒
            RedisCallback<String> callback = (connection) -> {
                JedisCommands commands = (JedisCommands) connection.getNativeConnection();
                return commands.set(key, clientId, "NX", "PX", expire);
            };
            String result = redisTemplate.execute(callback);

            return !StringUtils.isEmpty(result);
        } catch (Exception e) {
            logger.error("set redis occured an exception", e);
        }
        return false;
    }

    public String get(String key) {
        try {
            RedisCallback<String> callback = (connection) -> {
                JedisCommands commands = (JedisCommands) connection.getNativeConnection();
                return commands.get(key);
            };
            String result = redisTemplate.execute(callback);
            return result;
        } catch (Exception e) {
            logger.error("get redis occured an exception", e);
        }
        return "";
    }

    public boolean releaseLock(String key, String requestId) {
        // 释放锁的时候，有可能因为持锁之后方法执行时间大于锁的有效期，此时有可能已经被另外一个线程持有锁，所以不能直接删除
        try {
            List<String> keys = new ArrayList<>();
            keys.add(key);
            List<String> args = new ArrayList<>();
            args.add(requestId);

            // 使用lua脚本删除redis中匹配value的key，可以避免由于方法执行时间过长而redis锁自动过期失效的时候误删其他线程的锁
            // spring自带的执行脚本方法中，集群模式直接抛出不支持执行脚本的异常，所以只能拿到原redis的connection来执行脚本
            RedisCallback<Long> callback = (connection) -> {
                Object nativeConnection = connection.getNativeConnection();
                // 集群模式和单机模式虽然执行脚本的方法一样，但是没有共同的接口，所以只能分开执行
                // 集群模式
                if (nativeConnection instanceof JedisCluster) {
                    return (Long) ((JedisCluster) nativeConnection).eval(UNLOCK_LUA, keys, args);
                }

                // 单机模式
                else if (nativeConnection instanceof Jedis) {
                    return (Long) ((Jedis) nativeConnection).eval(UNLOCK_LUA, keys, args);
                }
                return 0L;
            };
            Long result = redisTemplate.execute(callback);

            return result != null && result > 0;
        } catch (Exception e) {
            logger.error("release lock occured an exception", e);
        } finally {
            // 清除掉ThreadLocal中的数据，避免内存溢出
            //lockFlag.remove();
        }
        return false;
    }

}

3.AOP切面：

import huayue.sports.venue.annotation.AvoidRepeatSubmit;
import huayue.sports.venue.common.BusinessException;
import lombok.extern.slf4j.Slf4j;
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.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;
import org.springframework.web.context.request.RequestContextHolder;
import org.springframework.web.context.request.ServletRequestAttributes;

import javax.servlet.http.HttpServletRequest;
import java.util.UUID;

/**
 * Created by yxh on 2022/1/10 18:00
 * 防止重复提交的切面
 */
@Aspect
@Component
@Slf4j
public class RepeatSubmitAspect {
    // 重复提交code
    private static final int REPEAT_SUBMIT_CODE = 507;

    @Autowired
    private RedisDistributedLock redisDistributedLock;


    /**
     * 切点
     *
     * @param avoidRepeatSubmit 注解
     */
    @Pointcut("@annotation(avoidRepeatSubmit)")
    public void pointCut(AvoidRepeatSubmit avoidRepeatSubmit) {
    }

    /**
     * 利用环绕通知进行处理重复提交问题
     *
     * @param pjp  ProceedingJoinPoint
     * @param avoidRepeatSubmit 注解
     * @return Object
     * @throws Throwable
     */
    @Around(value = "pointCut(avoidRepeatSubmit)", argNames = "pjp,avoidRepeatSubmit")
    public Object around(ProceedingJoinPoint pjp, AvoidRepeatSubmit avoidRepeatSubmit) throws Throwable {

        long lockMillSeconds = avoidRepeatSubmit.lockTime();

        //获得request对象
        HttpServletRequest request = httpServletRequest();

        Assert.notNull(request, "request can not null");

        // 此处可以用token
        String token = request.getParameter("access_token");
        String path = request.getServletPath();
        String key = getKey(token, path);
        log.info("key={}", key);
        String clientId = getClientId();
        //锁定多少毫秒
        boolean isSuccess = redisDistributedLock.setLock(key, clientId, lockMillSeconds);
        Object result;
        if (isSuccess) {
            log.info("tryLock success, key = [{}], clientId = [{}]", key, clientId);
            // 获取锁成功, 执行进程
            try {
                result = pjp.proceed();
            } finally {
                //解锁
                redisDistributedLock.releaseLock(key, clientId);
                log.info("releaseLock success, key = [{}], clientId = [{}]", key, clientId);
            }
            return result;
        } else {
            // 获取锁失败，认为是重复提交的请求
            log.info("tryLock fail, key = [{}]", key);
            throw new BusinessException("重复请求，请稍后再试");
            // return ResultUtils.fail(REPEAT_SUBMIT_CODE, "重复请求，请稍后再试");
        }
    }

    /**
     * 获得request对象
     *
     * @return HttpServletRequest对象
     */
    private HttpServletRequest httpServletRequest() {
        ServletRequestAttributes requestAttributes = (ServletRequestAttributes) RequestContextHolder.getRequestAttributes();
        assert requestAttributes != null;
        return requestAttributes.getRequest();
    }


    /**
     * 获得请求key
     *
     * @param token token
     * @param path  路径
     * @return 组合key
     */
    private String getKey(String token, String path) {
        return token + ":" + path;
    }

    /**
     * 获得uuid
     *
     * @return uuid
     */
    private String getClientId() {
        return UUID.randomUUID().toString();
    }


}

4.只要在需要防止重复提交的接口上面加上@AvoidRepeatSubmit 注解即可实现。

5.所需的依赖：

<dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
            <exclusions>
                <exclusion>
                    <groupId>io.lettuce</groupId>
                    <artifactId>lettuce-core</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

        <!-- https://mvnrepository.com/artifact/redis.clients/jedis -->
        <dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
            <version>2.9.0</version>
        </dependency>

Redis主从复制

概念

Redis的主从复制概念和MySQL的主从复制大概类似。一台主机master，一台从机slaver。master主机数据更新后根据配置和策略，自动同步到slaver从机，Master以写为主，Slave以读为主。

主要用途

• 读写分离：适用于读多写少的应用，增加多个从机，提高读的速度，提高程序并发

• 数据容灾恢复：从机复制主机的数据，相当于数据备份，如果主机数据丢失，那么可以通过从机存储的数据进行恢复。

• 高并发、高可用集群实现的基础：在高并发的场景下，就算主机挂了，从机可以进行主从切换，从机自动成为主机对外提供服务。

一主多从配置

环境准备

老哥太穷了，就用一台机器模拟三个机器。

• 第一步：将redis.conf复制3份，分别是redis6379.conf、redis6380.conf、redis6381.conf

• 第二步： 修改三个redis.conf文件里的port端口、pid文件名、日志文件名、rdb文件名

• 第三步： 分别打开三个窗口模拟三台服务器，并开启redis服务。

查看当前3台机器主从角色

先用命令info replication看看3台机器目前的角色是什么。

# 三台机器都是这个状态
127.0.0.1:6379> info replication
# 角色是master主机
role:master
# 从机个数为0
connected_slaves:0

设置主从关系

这里注意，我们只设置从机就可以了，不用设置主机。我们选择6380和6381作为从机。6379作为主机。

# 6380 端口
127.0.0.1:6380> SLAVEOF 127.0.0.1 6379

# 6381 端口
127.0.0.1:6381> SLAVEOF 127.0.0.1 6379

# 6381 端口
127.0.0.1:6381> SLAVEOF 127.0.0.1 6379

再次查看3台机器目前角色

再次执行命令：info replication

# 主机
127.0.0.1:6379> info replication
role:master # 角色：主机
connected_slaves:2 #连接的从机个数，以及从机IP和端口
slave0:ip=127.0.0.1,port=6380,state=online,offset=98,lag=1
slave1:ip=127.0.0.1,port=6381,state=online,offset=98,lag=1

# 从机1
127.0.0.1:6380> info replication
role:slave # 角色：从机
master_host:127.0.0.1 # 主机的IP和端口
master_port:6379

# 从机2
127.0.0.1:6381> info replication
role:slave # 角色：从机
master_host:127.0.0.1 # 主机的IP和端口
master_port:6379

搭建成功，试验一把

• 全量复制： 从机会把主机之前的数据全部都同步过来，大家可以在从机上get 某key试试。

• 增量复制： 当主机新增数据时，从机会将该新增数据同步过来，大家可以在主机上执行命令set key value，然后在从机上get 该key，看是否能获取到。

读写分离

Redis的从机默认不允许进行写操作，大家可以在从机上执行命令set key value，会报错。

# 6380从机
127.0.0.1:6380> set k3 v3
(error) READONLY You can't write against a read only slave.

「呼，好累」，主从复制写的差不多了！！

主从复制原理

全量复制

**「①」**slave发送psync，由于是第一次复制，不知道master的runid，自然也不知道offset，所以发送psync ？ -1

**「②」**master收到请求，发送master的runid和offset给从节点。

**「③」**从节点slave保存master的信息

**「④」**主节点bgsave保存rdb文件

**「⑤」**主机点发送rdb文件

并且在**「④」和「⑤」**的这个过程中产生的数据，会写到复制缓冲区repl_back_buffer之中去。

**「⑥」**主节点发送上面两个步骤产生的buffer到从节点slave

**「⑦」**从节点清空原来的数据，如果它之前有数据，那么久会清空数据

**「⑧」**从节点slave把rdb文件的数据装载进自身。

全量复制的开销

**「①」**bgsave时间

**「②」**rdb文件网络传输时间

**「③」**从节点清空数据的

**「④」**从节点加载rdb的时间

**「⑤」**可能的aof重写时间，这是针对从节点，例如开启了aof之后，从节点添加buffer数据时候，可能需要aof重写

基于上面的原因，有的情况下不适合使用全量复制，例如网络抖动之后，从节点只需要传送一部分数据，不需要传送全部数据，redis2.8之后实现了部分复制功能

部分复制

**「①」**假设发送网络抖动或者别的情况，暂时失去了连接

**「②」**这个时候，master还在继续往buffer里面写数据

**「③」**slave重新连接上了master

**「④」**slave向master发送自己的offset和runid

**「⑤」**master判断slave的offset是否在buffer的队列里面，如果是，那就返回continue给slave，否则需要进行全量复制（因为这说明已经错过了很多数据了）

**「⑥」**master发送从slave的offset开始到缓冲区队列结尾的数据给slave

1200页Java架构面试专题及答案

小编整理不易，对这份1200页Java架构面试专题及答案感兴趣劳烦帮忙转发/点赞一下，然后点击这里即可免费领取！

百度、字节、美团等大厂常见面试题

/点赞一下，然后点击这里即可免费领取！**

[外链图片转存中…(img-Jr4cT7sr-1627542919002)]

[外链图片转存中…(img-SAYxQ2Sz-1627542919003)]

百度、字节、美团等大厂常见面试题