多节点部署执行定时任务选举单一节点解决方案

水深无声 2022-05-09 01:56 347阅读 0赞

### **基本思路：** ###

利用缓存 + 随机数，定时任务首次执行时进行选举，并将选举结果保存下来，从而解决多节点定时任务重复跑的问题。任务分2个阶段，第一个阶段是投票阶段，第二个阶段是选举阶段。

**主要代码如下**：

*/\*\**

*     \* 首次选举后不再选举*

*     \**

*     \* @param key*

*     \* @param randomTime*

*     \* @return*

*     \* @author xx2018-9-26*

*     \*/*

@Deprecated

*    public boolean getExecFirstAth(String key, Integer randomTime) \{*

*        boolean getExecAth = false;*

*        if (execFlag) \{*

*            getExecAth = this.getExecAth(key, randomTime);*

*        \}*

*        if (getExecAth) \{*

*            retFlag = true;*

*        \}*

*        execFlag = false;*

*        log.info("首次选举后不再选举 （true 抢到权限， false 没有抢到） -- ：\{\}",retFlag);*

*        return retFlag;*

*    \}*

*    /\*\**

*     \* 权限分配方法，抢到机会返回true，否则返回false*

*     \* 仅针对于 定时任务同时启动的场景*

*     \**

*     \* @param key        建议命名规范：项目名\_执行内容 如 xx\_admin\_importUser*

*     \* @param randomTime 随机时间，越长重叠概率越小*

*     \* @return*

*     \* @author xx2018-9-26*

*     \*/*

*    public boolean getExecAth(String key, Integer randomTime) \{*

*        //1 释放缓存*

*        cacheUtil.putHazelCastValue(key, "");*

*        //2 放入数据*

*        Random random = new Random();*

*        Integer time = random.nextInt(randomTime);*

*        log.debug("time -- > \{\}", time);*

*        MarketUtil.sleepThread(time);*

*        setIntoHazelcast(key, time);*

*        Set<Integer> getExecAth;*

*        //3 校验放入结果*

*        boolean breakFlag = true;*

*        while (breakFlag) \{*

*            getExecAth = JsonUtil.jsonToObject(cacheUtil.gettHazelCastValue(key), HashSet.class);*

*            if (getExecAth.contains(time)) \{*

*                breakFlag = false;*

*                log.debug("放入缓存成功！");*

*            \} else \{*

*                setIntoHazelcast(key, time);*

*            \}*

*        \}*

*        MarketUtil.sleepThread(randomTime);*

*        //4 取出比较*

*        getExecAth = JsonUtil.jsonToObject(cacheUtil.gettHazelCastValue(key), HashSet.class);*

*        log.debug("getExecAth -- > \{\}", getExecAth);*

*        boolean flag = false;*

*        if (!CollectionUtils.isEmpty(getExecAth)) \{*

*            Integer max = getExecAth.stream().max(Integer::compare).get();*

*            flag = max.equals(time);*

*        \}*

*        log.info("获取权限返回值（true 抢到权限， false 没有抢到） -- ：\{\}", flag);*

*        return flag;*

*\}*

/**
     * 设置缓存值
     *
     * @param key
     * @param time
     * @author xx 2018-8-14
     */
    private void setIntoHazelcast(String key, Integer time) {
        Set<Integer> execAth = new HashSet<>();
        execAth.add(time);
        Set<Integer> getExecAth = JsonUtil.jsonToObject(cacheUtil.gettHazelCastValue(key), HashSet.class);
        if (!CollectionUtils.isEmpty(getExecAth)) {
            log.debug("未取到，放入缓存！");
            getExecAth.addAll(execAth);
            cacheUtil.putHazelCastValue(key, JsonUtil.objectToJson(getExecAth));
        } else {
            cacheUtil.putHazelCastValue(key, JsonUtil.objectToJson(execAth));
        }
    }

调用方法：

在定时任务的第一行加入如下代码即可

//定义一个选举机制
    if (!xxxUtil.getExecFirstAth("xxxxx_xxxxxxxx", 60000)) {
        return;
    }

*说明： getExecFirstAth 方法是对getExecAth 方法的延申，即增加了首次选举结果保存功能，这样每次启动服务，只会选举一次，后面不用选举，保证日志在一个服务里，好定位问题。（失败概率的问题，可以忽略，非常非常非常小）*

* **但是后来发现这个方法存在漏洞，对于节点重启或者节点缩容/扩容的情况处理的不好，因此getExecFirstAth 方法，暂时废弃**。还是每次都选举的方式稳定。*

**温馨提示：*****一定注意把缓存的命名空间，主键key值改成自己的，节点扩展时，要重启多有服务***