Curator学习笔记（三） - zk防重复提交2.0版

写一点笔记

发布于 2022-08-11 15:45:01

2250

发布于 2022-08-11 15:45:01

文章被收录于专栏：程序员备忘录

前几天，说要用curator的读写锁写一个分布式防重复提交的工具包。然后今天作者就探索一下，在上次文章的末尾，作者说当时的这种方式解决不了大量表单使用相同的防重复提交token上送。也就是比如网站的首页要加载很多请求，然后这些请求还都适用相同的token，所以说这些使用相同token的请求只能通过一个。其他都可能被视为重复的表单。但是作者后边想了想感觉这种情况还是比较少。很多时候表单都是单个提交的。所以我们先不考虑那种情况，因为如果考虑那种情况会比较复杂。这里我们还是以单个表单作为示例写一个注解来做这件事情。

做出来的效果就是这样的：

类说明：

CuratorConfig 表示curator的客户端。交由spring管理
DoubleSubmitAdvice 是对请求防重复提交token的生成处理。
ReSubmitLock 是使用读写锁进行的分布式token校验的核心处理类
DoubleSubmitAnnotation 是注解，该注解只能用着@Controller和@RestController下的方法上。
DoubleSubmitAsject 是使用Spring Aop定义的切面。
Testlock 是一个测试类。

效果是可以通过注解上的策略来生成或者校验token的有效性。


@DoubleSubmitAnnotation(check = true,generate = true)
@GetMapping(value = "/test")
public ResponseResult test() {
     return ResponseResult.success(123);
 }
postMan测试一下就是这样的。

注解：

@Documented
@Inherited
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
public @interface DoubleSubmitAnnotation {



    /**
     * 是否校验token，默认为false
     * @return
     */
    boolean check() default false;


    /**
     * 是否生成新的token
     * @return
     */
    boolean generate() default false;
}

切面：

@Aspect
@Configuration
public class DoubleSubmitAsject {

    private Logger logger = LoggerFactory.getLogger(DoubleSubmitAsject.class);
    /**
     * 进行一些操作
     */
    @Autowired
    private ReSubmitLock reSubmitLock;


    @Value("${double.submit.token:ztoken}")
    private String tokenName;

    /**
     * /**
     * 设置操作异常切入点记录异常日志 扫描所有controller包下操作
     */
    @Pointcut("@annotation(com.scaffold.simple.admin.lock.annotation.DoubleSubmitAnnotation)")
    public void doubleSubmit() {
    }

    /**
     * 执行之前进行
     *
     * @param joinPoint 切点
     */
    @Before(value = "doubleSubmit()")
    public void doBefore(JoinPoint joinPoint) throws Exception {
        boolean isRestController = joinPoint.getTarget().getClass().isAnnotationPresent(RestController.class);
        boolean isController = joinPoint.getTarget().getClass().isAnnotationPresent(Controller.class);
        if (isRestController | isController) {
            ServletRequestAttributes attributes = (ServletRequestAttributes) RequestContextHolder.getRequestAttributes();
            assert attributes != null;
            String token = attributes.getRequest().getHeader(tokenName);
            String userId = SessionUtils.getUserId();
            MethodSignature methodSignature = (MethodSignature) joinPoint.getSignature();
            Method method = methodSignature.getMethod();
            DoubleSubmitAnnotation doubleSubmitAnnotation = method.getAnnotation(DoubleSubmitAnnotation.class);
            if (!StringUtils.isEmpty(token) && !StringUtils.isEmpty(userId)) {

                if (!Objects.isNull(doubleSubmitAnnotation)) {
                    if (doubleSubmitAnnotation.check()) {
                        if (!reSubmitLock.check(userId, token)) {
                            logger.warn(MessageFormat.format("重复提交的表单：{0}：{1}", userId, token));
                            throw new Exception("重复的表单，请重新提交");
                        }
                    }
                }
            } else {
                logger.warn(MessageFormat.format("无效的放重复检验：{0}：{1}", userId, token));
            }
            if (doubleSubmitAnnotation.generate()) {
                SessionUtils.setSubmitToken(reSubmitLock.generateToken(userId));
            }
        }
    }

    /**
     * 返回打印日志
     *
     * @param joinPoint 切点
     * @param keys      返回的数据
     */
    @AfterReturning(value = "doubleSubmit()", returning = "keys")
    public void saveOperLog(JoinPoint joinPoint, Object keys) {
    }
}

curator客户端配置

@Configuration
public class CuratorClientConfig {


    @Bean
    public CuratorFramework main() {    
   //        每3秒重连一次，重连3次
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        //创建连接对象
        CuratorFramework client = CuratorFrameworkFactory.builder()
                //IP地址端口号，集群模式
                .connectString("127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183")
                //客户端与服务器之间的会话超时时间
                .sessionTimeoutMs(10000)
                //当客户端与服务器之间会话超时3s后，进行一次重连
                .retryPolicy(retryPolicy)
                //命名空间，当我们创建节点的时候，以/create为父节点
                .namespace("create")
                //构建连接对象
                .build();
        //打开连接
        client.start();
        //是否成功建立连接，true ：建立， false：没有建立
        System.out.println(client.isStarted());
        return client;
    }
}

对token的后置处理

@ControllerAdvice
public class DoubleSubmitAdvice implements ResponseBodyAdvice{


    @Override
    public boolean supports(MethodParameter returnType, Class<? extends HttpMessageConverter> converterType) {
        return true;
    }


    @Override
    public Object beforeBodyWrite(Object body, MethodParameter returnType, MediaType selectedContentType, Class<? extends HttpMessageConverter> selectedConverterType, ServerHttpRequest request, ServerHttpResponse response) {
        if (!StringUtils.isEmpty(SessionUtils.getSubmitToken())){
            //token有效 则将token放入cookie中
            Cookie tokenCookie = new Cookie("ztoken", SessionUtils.getSubmitToken());
            tokenCookie.setPath("/");
            tokenCookie.setDomain("localhost");
            // 会话级cookie，关闭浏览器失效
            tokenCookie.setMaxAge(-1);
            ServletServerHttpResponse resp = (ServletServerHttpResponse)response;
            resp.getServletResponse().addCookie(tokenCookie);
        }
        return body;
    }
}

token的校验逻辑和token生成

@Component
public class ReSubmitLock {
    /**
     * 进行一些操作
     */
    @Autowired
    private CuratorFramework client;

    /**
     * 放重复提交
     * @param key
     * @param token
     * @return
     * @throws Exception
     */
    public boolean check(String key, String token) throws Exception {
        boolean status = false;
        String mainKey="/"+key;
        InterProcessReadWriteLock interProcessReadWriteLock = new InterProcessReadWriteLock(client, mainKey);
        InterProcessLock interProcessLock = interProcessReadWriteLock.writeLock();
        System.out.println("等待获取锁对象!");
        // 获取锁
        try {
            interProcessLock.acquire();
            // 读取数据时读取节点的属性
            Stat stat = new Stat();
            byte[] zkToken = client.getData()
                    .storingStatIn(stat)
                    .forPath(mainKey);
            String oldToken = new String(zkToken);
            System.out.println("允许的Token：" + oldToken);
            if (token.equals(oldToken)) {
                System.out.println("校验成功！");
                generateToken(key);
                status = true;
            }
        } catch (Exception e){
            e.printStackTrace();
            System.out.println("节点加锁产生错误");
        }finally {
            // 释放锁
            interProcessLock.release();
            System.out.println("等待释放锁!");
        }
        if (!status) {
            System.out.println("不能重复提交表单");
        }
        return status;
    }

    /**
     * 创建token
     */
    public String generateToken(String key) throws Exception {
        String mainKey="/"+key;
        String newToken = UUID.randomUUID().toString();
        System.out.println("设置的新token为：" + newToken);
        // 判断节点是否存在,为null表示不存在
        Stat stat= client.checkExists()
                // 节点路径
                .forPath(mainKey);
        if (!Objects.isNull(stat)){
            client.setData()
                    .forPath(mainKey, newToken.getBytes());
        }else{
            client.create()
                    .forPath(mainKey, newToken.getBytes());
        }
        return newToken;
    }
}

注解的使用：

@RestController
@RequestMapping(value = "/zklock")
public class TestLock {

    /**
     * 放重复提交
     */
    @Autowired
    private ReSubmitLock reSubmitLock;


    /**
     * 校验并生成下次凭证
     * @return
     */
    @DoubleSubmitAnnotation(check = true,generate = true)
    @GetMapping(value = "/test")
    public ResponseResult test() {
        return ResponseResult.success(123);
    }

    /**
     * 生成新token
     * @return
     */
    @DoubleSubmitAnnotation(generate = true)
    @GetMapping(value = "/gen")
    public ResponseResult test1() {
        return ResponseResult.success(123);
    }
}

相关代码已经提交到github，欢迎提bug！，以后有相关需求的时候咋再打成starter。先这样吧。

https://github.com/tianjingle/simple-admin

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2021-02-04，如有侵权请联系 cloudcommunity@tencent.com 删除

github