分布式场景下后端实现多机器树节点的同步

曾经终败给现在 2022-09-14 14:27 19阅读 0赞

## 一、需求背景 ##

![组织树][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAaHBfeWFuZ3Blbmc_size_17_color_FFFFFF_t_70_g_se_x_16_pic_center]

如上图所示：  
背景知识：组织树中的节点信息是保存到各台服务器的内存中的；  
需求：组织树中需要添加新的节点，添加完成后，所有服务器内存中组织树的数据是完全一样的；

## 二、实现方案 ##

通过redis分布式锁的方式实现上述功能；

### 1、记录对当前节点的修改 ###

当前节点的修改包括：增加节点、修改节点、删除、复制、导入、批量删除、批量更新等操作；

List<Kamala2NodeSynLogDO> nodeSynLogDOS = this.commonGetDOByIp(ip -> {
                    Kamala2NodeSynLogDO nodeSynLogDO = this.initDO(caseNodeTypeEnum, Kamala2NodeLogTypeEnum.UPDATE);
                    if (ServiceRedisUtils.getAddr().equals(ip)) {
                        nodeSynLogDO.setStatus(Kamala2NodeLogStatusEnum.FINISHED);
                    }
                    //必填三项
                    nodeSynLogDO.setConsumeServiceId(ip);
                    nodeSynLogDO.setNodeTypeEnum(caseNodeTypeEnum);
                    nodeSynLogDO.setNodeField(Kamala2NodeLogFieldEnum.COUNT);
                    nodeSynLogDO.setZipNodeJson(count.toString());
                    nodeSynLogDO.setNodeId(treeNodeDTO2.getKey());
                    return nodeSynLogDO;
                });

### 2、将操作信息保存到数据库中 ###

//保存到数据库中的数据，包含集群中所有节点的IP信息及当前Ip对应的同步状态；即除了当前机器外，集群内的其他机器均处于init的状态；
    ResultModel<List<Kamala2NodeSynLogDO>> resultModel = kamala2NodeSynLogService.saveOrUpdateBatch(kamala2NodeSynLogDOs);

### 3、加锁，并将操作放到redis队列中； ###

jedisService.lock(lockKey);
    map.forEach((k, v) -> {
              jedisService.putList(LOG_LIST_PREFIX + k, v.toString(), false);
     });
    logger.info("节点日志存入队列,services:{}", JSONObject.toJSONString(map.keySet()));

### 4、解锁 ###

jedisService.unLock(lockKey);

### 5、创建定时任务，每隔1s查询redis队列中是否有数据，如果有，则详细查询数据库，基于数据库中的结果进行相关的增删改查操作； ###

@Scheduled(fixedRate = 1000)
        public void getNodeLog(){
            kamala2NodeSynLogConsumer.run();
        }

run方法中部分逻辑如下：

//判断队列中是否有数据，如果有，则进行相关的操作
    String idStr = jedisService.popList(logListKey, true, String.class);
    if (StringUtils.isNotBlank(idStr)){
    //基于当前服务器的IP来查询数据库中的数据是否处于finish状态，如果为是，则表示当前机器已经完成内存中节点的同步操作；
    //否则就基于数据库中的数据当前服务器的节点进行增删改查操作；
    }

lock代码如下：

public void lock(@NotNull String lockKey, long expireTime) {
            if (expireTime == 0L) {
                expireTime = SYN_TIME_OUT;
            }
            long threadId = Thread.currentThread().getId();
            long size = expireTime / SYN_SPACE_TIME;
            size = size <= 0 ? 1 : size;
            for (int i = 0; i < size; i++) {
                if (!this.getLock(lockKey, expireTime, threadId)) {
                    try {
                        Thread.sleep(SYN_SPACE_TIME);
                    } catch (InterruptedException e) {
                        logger.error("lock sleep is error, error is {}", ExceptionUtils.getStackTrace(e));
                    }
                } else {
                    return;
                }
            }
            throw new RedisLockException("lock get is time out");
        }
        //判断是否能获取到锁；
    private boolean getLock(String lockKey, long expireTime, long threadId) {
            try (Jedis jedis = jedisPool.getResource()) {
                String uuid = UUIDUtil.generateUUID();
                String requestId = KEY_MAP_REQUEST_ID.putIfAbsent(lockKey, uuid);
                if (StringUtils.isBlank(requestId)) {
                    requestId = uuid + ":" + threadId;
                } else {
                    requestId = requestId + ":" + threadId;
                }
                SetParams setParams = new SetParams();
                setParams.nx().px(expireTime);
                String result = jedis.set(lockKey, requestId, setParams);
                return LOCK_SUCCESS.equals(result);
            } catch (Exception e) {
                logger.error("get lock is error , error is {}", ExceptionUtils.getStackTrace(e));
                return false;
            }
        }

unlock代码如下：

public void unLock(@NotNull String lockKey) {
            long threadId = Thread.currentThread().getId();
            String requestId = KEY_MAP_REQUEST_ID.get(lockKey);
            if (StringUtils.isBlank(requestId)) {
                throw new RedisLockException("release lock is error , lock no local service");
            }
            requestId = requestId + ":" + threadId;
            Object result = null;
            try (Jedis jedis = jedisPool.getResource()) {
                result = jedis.eval(LUA_SCRIPT, Collections.singletonList(lockKey), Collections.singletonList(requestId));
            } catch (Exception e) {
                logger.error("release lock is error , error is {}", ExceptionUtils.getStackTrace(e));
            }
            if (!RELEASE_SUCCESS.equals(result)) {
                throw new RedisLockException("release lock is error , lock no Thread or lock is time out");
            } else {
                KEY_MAP_REQUEST_ID.remove(lockKey);
            }
        }

同时附上redis 分布式锁实现文章链接：https://www.jianshu.com/p/47fd7f86c848

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAaHBfeWFuZ3Blbmc_size_17_color_FFFFFF_t_70_g_se_x_16_pic_center]: /images/20220828/b5b5ece96fbc4b0d913c2af9b975b020.png