自定义注解解决消息幂等性问题

1.什么是幂等性问题

        用户的一个请求由于网络波动未完成，而用户那边并不知道，于是反复点击刷新，这时候后端就会接收到大量相同的请求，这时候就需要进行幂等处理，使得多个相同请求跟一个请求所得结果相同。而对于get和delete请求，具有天生的幂等性，故只用考虑put和post请求即可。还有一种情况则是消息队列的重复消费问题，其原因是一样的。

2.如何解决

        利用token的验证机制，当第一个请求过来时，向redis中写入一个key-value，其中key需要保证相同请求一定相同，不同请求则一定不同，来保证请求是相同的，当然，多个相同的请求也可能是由于订阅模式造成的，所以还需要根据value来判断是否是同一用户发起的请求。

3代码实现

定义注解

@Target({ElementType.TYPE,ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface Repeat {
    //消息的最长消费时间，超过该时间则判断消费失败
    long time() default 60000L;
}

定义切面

        其中RepeatServiceAdapter是一个接口类，可以自定义对什么类型的消息进行幂等处理，这里只给出处理http请求的重复请求问题，后续添加处理器只需要添加实现类即可

@Aspect
@Component
public class RepeatAspect {
    @Resource
    private List<RepeatServiceAdapter> repeatServiceAdapters;//自动注入接口的所有实现类
 
    @Pointcut("@annotation(com.cg.annotation.Repeat)")
    public void pointcut() {
    }
 
    @Around("pointcut()")
    public Object doAround(ProceedingJoinPoint pjp) throws Throwable {
        //获取代理的方法
        MethodSignature methodSignature = (MethodSignature) pjp.getSignature();
        Method method = methodSignature.getMethod();
        for (RepeatServiceAdapter repeatServiceAdapter : repeatServiceAdapters) {
            //判断是否符合对应的设配器
            if (repeatServiceAdapter.support(method) || repeatServiceAdapter.support(method.getClass())) {
                return repeatServiceAdapter.resolve(pjp);
            }
        }
        //没有适配的则直接放行
        return pjp.proceed();
    }
}

定义处理逻辑的适配器

/**
 * 解决重复请求的问题
 */
public interface RepeatServiceAdapter {
    /**
     * 判断是否符合该适配器
     * @param clazz
     * @return
     */
    boolean support(Class<?> clazz);
 
    boolean support(Method method);
    /**
     * 处理重复请求
     * @param pjp 切面类
     * @return 处理结果
     * @throws Throwable
     */
    Object resolve(ProceedingJoinPoint pjp) throws Throwable;
 
}

/**
 * 处理http重复请求问题
 */
@Component
@Slf4j
public class RequestRepeatAdapter implements RepeatServiceAdapter {
    @Autowired
    private RedisTemplate<String, String> redisTemplate;
 
    @Override
    public boolean support(Class<?> clazz) {
        return clazz.isAnnotationPresent(PostMapping.class) || clazz.isAnnotationPresent(PutMapping.class);
    }
 
    @Override
    public boolean support(Method method) {
        return method.isAnnotationPresent(PostMapping.class) || method.isAnnotationPresent(PutMapping.class);
    }
 
    @Override
    public Object resolve(ProceedingJoinPoint pjp) throws Throwable {
        MethodSignature methodSignature = (MethodSignature) pjp.getSignature();
        //获取方法上的注释
        Method method = methodSignature.getMethod();
        //获取request
        ServletRequestAttributes attributes = (ServletRequestAttributes) RequestContextHolder.getRequestAttributes();
        //获取请求头中的repeat属性
        HttpServletRequest request = attributes.getRequest();
        //repeat是前端访问时需要携带的key值
        String repeat = request.getHeader("repeat");
        //获取请求路径中的userId,也可以从token中解析
        String userId = request.getParameter("userId");
        //判断是否存在该请求头的uuid以及对应的userId
        return getObject(pjp, method, repeat, userId);
    }
 
    private Object getObject(ProceedingJoinPoint pjp, Method method, String repeat, String userId) throws Throwable {
        if (Boolean.TRUE.equals(redisTemplate.hasKey(repeat)) && userId.equals(redisTemplate.opsForValue().get(repeat))) {
            //判断该请求为重复请求，执行拦截
            log.info("重复请求...");
            return ResultInfo.error(CodeEnum.REPEAT_REQUEST);
        } else {
            //不是重复请求，则放行
            //获取最长重复等待时间并保存uuid
            Repeat annotation = method.getAnnotation(Repeat.class);
            long time = annotation.time();
            //双重检查锁保证线程安全
            synchronized (this.getClass()) {
                if (Boolean.TRUE.equals(redisTemplate.hasKey(repeat)) && userId.equals(redisTemplate.opsForValue().get(repeat))) {
                    //判断该请求为重复请求，执行拦截
                    log.info("重复请求...");
                    return ResultInfo.error(CodeEnum.REPEAT_REQUEST);
                }
                redisTemplate.opsForValue().set(repeat, userId, time, TimeUnit.MILLISECONDS);
            }
            log.info("执行请求...");
            Object proceed = pjp.proceed();
            //执行结束后删除uuid
            redisTemplate.delete(repeat);
            return proceed;
        }
    }
}

4.结果验证

controller方法

    @Repeat
    @PostMapping("login")
    public ResultInfo<?> insert() throws InterruptedException {
        testMapper.insert();
        //模拟网络波动
        Thread.sleep(1000);
        return ResultInfo.success(CodeEnum.SUCCESS);
    }

多线程模拟重复请求

public class Client {
    @Test
    public void test() throws Exception{
        for (int i = 0; i < 10; i++) {
            new Thread(()->{
                try {
                    String url = "http://localhost:8010/login?userId=1"; // 要发送POST请求的URL
                    URL obj = new URL(url);
                    HttpURLConnection con = (HttpURLConnection) obj.openConnection();
                    con.setRequestMethod("POST");
                    con.setRequestProperty("repeat", "b5ed28e9-e7c2-4a05-8b4d-aaee58984e24"); // 在HTTP头中添加名为"repeat"的字段
                    BufferedReader in = new BufferedReader(new InputStreamReader(con.getInputStream()));
                    String inputLine;
                    StringBuffer response = new StringBuffer();
                    while ((inputLine = in.readLine()) != null) {
                        response.append(inputLine);
                    }
                    in.close();
                    String json = response.toString(); // JSON数据
                    System.out.println(json);
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            }).start();
        }
        Thread.sleep(100000L);
    }
}

结果