更新居民 类别数据 和 更新房屋内居民 使用定时任务同步

epmet-commons/epmet-commons-tools/src/main/java/com/epmet/commons/tools/redis/RedisKeys.java

@@ -803,4 +803,18 @@ public class RedisKeys {
 	}
 
 
+	/**
+	 * desc: 数据同步业务key 根据方法名区分
+	 *
+	 * @param bizType [resi]
+	 * @return java.lang.String
+	 * @author LiuJanJun
+	 * @date 2022/7/16 7:12 下午
+	 */
+	public static String getDataSyncKey(String bizType) {
+		if (StringUtils.isBlank(bizType)){
+			throw new EpmetException(EpmetErrorCode.EPMET_COMMON_OPERATION_FAIL.getCode(),"参数错误");
+		}
+		return rootPrefix.concat("datasync:").concat(bizType);
+	}
 }

epmet-commons/epmet-commons-tools/src/main/java/com/epmet/commons/tools/redis/RedisUtils.java

@@ -357,6 +357,24 @@ public class RedisUtils {
         return redisTemplate.opsForSet().members(key);
     }
 
+    /**
+     * @Description 获取一个（set）
+     * @param key
+     * @author zxc
+     */
+    public Object sPop(String key){
+        return redisTemplate.opsForSet().pop(key);
+    }
+
+    /**
+     * @Description 添加一个元素（set）
+     * @param key
+     * @author zxc
+     */
+    public Long sAdd(String key,Object value){
+        return redisTemplate.opsForSet().add(key,value);
+    }
+
     /**
      * @Description  获取 zset 中带score的集合
      * @param key

epmet-module/epmet-job/epmet-job-server/src/main/java/com/epmet/task/ic/IcDataSyncTask.java

@@ -0,0 +1,36 @@
+package com.epmet.task.ic;
+
+import com.epmet.commons.tools.utils.Result;
+import com.epmet.feign.EpmetUserOpenFeignClient;
+import com.epmet.task.ITask;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+
+/**
+ * desc:定时弥补数据 保证数据的一致性
+ */
+@Slf4j
+@Component("icDataSyncTask")
+public class IcDataSyncTask implements ITask {
+
+    @Resource
+    private EpmetUserOpenFeignClient epmetUserOpenFeignClient;
+
+    /**
+     * 执行定时任务接口
+     *
+     * @param params 参数，多参数使用JSON数据
+     */
+    @Override
+    public void run(String customerId) {
+        log.info("icDataSyncTask定时任务正在执行，参数为customerId：{}", customerId);
+        Result result = epmetUserOpenFeignClient.icDataSyncTask(customerId);
+        if (result.success()) {
+            log.info("icDataSyncTask定时任务正在执行定时任务执行成功");
+        } else {
+            log.warn("icDataSyncTask定时任务正在执行定时任务执行失败:" + result.getMsg());
+        }
+    }
+}

epmet-user/epmet-user-client/src/main/java/com/epmet/feign/EpmetUserOpenFeignClient.java

@@ -835,4 +835,15 @@ public interface EpmetUserOpenFeignClient {
      */
     @PostMapping("/epmetuser/icresiuser/getIcUserCount")
     Result<Map<String, Integer>> getIcUserCount(@RequestBody IcUserCountFormDTO formDTO);
+
+    /**
+     * desc: 同步数据，房内居民数量 类别统计数据量等只用客户id当条件
+     *
+     * @param customerId
+     * @return com.epmet.commons.tools.utils.Result<java.lang.String>
+     * @author LiuJanJun
+     * @date 2022/7/16 6:53 下午
+     */
+    @PostMapping("/epmetuser/icResiUserDataSync/icDataSyncTask")
+    Result<Boolean> icDataSyncTask(String customerId);
 }

epmet-user/epmet-user-client/src/main/java/com/epmet/feign/fallback/EpmetUserOpenFeignClientFallback.java

@@ -636,4 +636,9 @@ public class EpmetUserOpenFeignClientFallback implements EpmetUserOpenFeignClien
         return ModuleUtils.feignConError(ServiceConstant.EPMET_USER_SERVER, "getIcUserCount", formDTO);
     }
 
+    @Override
+    public Result<Boolean> icDataSyncTask(String customerId) {
+        return ModuleUtils.feignConError(ServiceConstant.EPMET_USER_SERVER, "icDataSyncTask", customerId);
+    }
+
 }

epmet-user/epmet-user-server/src/main/java/com/epmet/controller/IcResiUserDataSyncController.java

@@ -0,0 +1,101 @@
+/**
+ * Copyright 2018 人人开源 https://www.renren.io
+ * <p>
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ * <p>
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ * <p>
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package com.epmet.controller;
+
+import com.epmet.commons.tools.distributedlock.DistributedLock;
+import com.epmet.commons.tools.feign.ResultDataResolver;
+import com.epmet.commons.tools.redis.RedisKeys;
+import com.epmet.commons.tools.redis.RedisUtils;
+import com.epmet.commons.tools.utils.Result;
+import com.epmet.service.IcResiUserDataSyncService;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.redisson.api.RLock;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+
+/**
+ * 用户基础信息同步计算入口
+ *
+ * @author generator generator@elink-cn.com
+ * @since v1.0.0 2021-10-26
+ */
+@Slf4j
+@RestController
+@RequestMapping("icResiUserDataSync")
+public class IcResiUserDataSyncController implements ResultDataResolver {
+
+    @Autowired
+    private IcResiUserDataSyncService icResiUserDataSyncService;
+    @Autowired
+    private RedisUtils redisUtils;
+    @Autowired
+    private DistributedLock distributedLock;
+
+
+    /**
+     * 获取录入居民的数量
+     *
+     * @Param formDTO
+     * @Return {@link Result<Map<String, Integer>>}
+     * @Author zhaoqifeng
+     * @Date 2022/6/30 9:35
+     */
+    @PostMapping("icDataSyncTask")
+    public Result<Boolean> icDataSyncTask(@RequestParam("customerId") String customerId) {
+        RLock lock = null;
+        String key = RedisKeys.getDataSyncKey("resi");
+        try {
+            lock = distributedLock.getLock(RedisKeys.getLockByMethodName("icDataSyncTask"),
+                    30L, 30L, TimeUnit.SECONDS);
+            if (StringUtils.isBlank(customerId)) {
+                Object customerIdObj = redisUtils.sPop(key);
+                while (customerIdObj != null) {
+                    try {
+                        icResiUserDataSyncService.icDataSyncTask(customerIdObj.toString());
+                        log.debug("icDataSyncTask customerId:{} execute success", customerIdObj);
+
+                        customerIdObj = redisUtils.sPop(key);
+                    } catch (Exception e) {
+                        log.error("icDataSyncTask exception msg:{},customerId:{}", e.getMessage(), customerIdObj);
+                        redisUtils.sAdd(key, customerIdObj);
+                    }
+                    log.debug("icDataSyncTask finish customerId:{} execute success", customerIdObj);
+                }
+            } else {
+                icResiUserDataSyncService.icDataSyncTask(customerId);
+                log.debug("icDataSyncTask customerId:{} execute success", customerId);
+            }
+        } catch (Exception e) {
+            if (StringUtils.isNotBlank(customerId)) {
+                redisUtils.sAdd(key, customerId);
+            }
+            log.error("icDataSyncTask exception msg:{},customerId:{}", e.getMessage(), customerId);
+        } finally {
+            distributedLock.unLock(lock);
+        }
+        return new Result<Boolean>().ok(true);
+    }
+}

epmet-user/epmet-user-server/src/main/java/com/epmet/mq/listener/ICWarnStatsEventListener.java

@@ -70,9 +70,11 @@ public class ICWarnStatsEventListener implements MessageListenerConcurrently {
         IcResiUserAddMQMsg obj = JSON.parseObject(msg, IcResiUserAddMQMsg.class);
         log.info("obj is {}", JSON.toJSONString(obj));
         try {
+            String key = RedisKeys.getDataSyncKey("resi");
+            redisUtils.sAdd(key, obj.getCustomerId());
             //todo 统计预警人数和 更新屋内居民数 用定时任务跑吧 这样太费劲
-            //||!updateIcHouseResiNumber(obj)
+            //||!updateIcHouseResiNumber(obj)  !resiWarnStats(obj)||  installed of IcDataSyncTask
-            if (!resiWarnStats(obj)||!updateIcNatResiFlag(tags, obj)){
+            if (!updateIcNatResiFlag(tags, obj)){
                 throw new EpmetException(EpmetErrorCode.EPMET_COMMON_OPERATION_FAIL.getCode(),"消费失败,等待重新消费");
             }
         } catch (RenException e) {

epmet-user/epmet-user-server/src/main/java/com/epmet/service/IcResiUserDataSyncService.java

@@ -0,0 +1,36 @@
+/**
+ * Copyright 2018 人人开源 https://www.renren.io
+ * <p>
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ * <p>
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ * <p>
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package com.epmet.service;
+
+/**
+ * 用户基础信息
+ *
+ * @author generator generator@elink-cn.com
+ * @since v1.0.0 2021-10-26
+ */
+public interface IcResiUserDataSyncService {
+    /**
+     * desc: 同步与居民相关的数据
+     *
+     * @param customerId
+     * @return java.lang.String
+     * @author LiuJanJun
+     * @date 2022/7/16 8:06 下午
+     */
+    Boolean icDataSyncTask(String customerId);
+}

epmet-user/epmet-user-server/src/main/java/com/epmet/service/IcResiUserService.java

@@ -490,4 +490,5 @@ public interface IcResiUserService extends BaseService<IcResiUserEntity> {
      */
     Map<String, Integer> getIcUserCount(IcUserCountFormDTO formDTO);
 
+    String icDataSyncTask(String customerId);
 }

epmet-user/epmet-user-server/src/main/java/com/epmet/service/impl/IcResiUserDataSyncServiceImpl.java

@@ -0,0 +1,55 @@
+package com.epmet.service.impl;
+
+import com.epmet.service.IcResiUserDataSyncService;
+import com.epmet.service.IcResiUserService;
+import com.epmet.service.StatsResiWarnService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import java.util.concurrent.ExecutorService;
+
+/**
+ * desc:
+ *
+ * @author: LiuJanJun
+ * @date: 2022/7/16 8:07 下午
+ * @version: 1.0
+ */
+@Slf4j
+@Service
+public class IcResiUserDataSyncServiceImpl implements IcResiUserDataSyncService {
+
+    @Autowired
+    private IcResiUserService icResiUserService;
+    @Autowired
+    private StatsResiWarnService statsResiWarnService;
+    @Autowired
+    private ExecutorService executorService;
+
+    /**
+     * desc: 同步与居民相关的数据
+     *
+     * @param customerId
+     * @return java.lang.String
+     * @author LiuJanJun
+     * @date 2022/7/16 8:06 下午
+     */
+    @Override
+    public Boolean icDataSyncTask(String customerId) {
+        Boolean flag = true;
+        try {
+            statsResiWarnService.resiWarn(customerId);
+        } catch (Exception e) {
+            flag = false;
+            log.error("icDataSyncTask resiWarn exception:{}", e.getMessage(), customerId);
+        }
+        try {
+            icResiUserService.updateIcHouseResiNumber(customerId);
+        } catch (Exception e) {
+            flag = false;
+            log.error("icDataSyncTask updateIcHouseResiNumber exception:{}", e.getMessage(), customerId);
+        }
+        return flag;
+    }
+}

epmet-user/epmet-user-server/src/main/java/com/epmet/service/impl/IcResiUserServiceImpl.java

@@ -2763,6 +2763,12 @@ public class IcResiUserServiceImpl extends BaseServiceImpl<IcResiUserDao, IcResi
         return list.stream().collect(Collectors.toMap(IcUserCountResultDTO::getId, IcUserCountResultDTO::getCount, (key1 , key2)-> key2));
     }
 
+    @Override
+    public String icDataSyncTask(String customerId) {
+        operCustomizeOpenFeignClient
+        return null;
+    }
+
     /**
      * 根据身份证获取居民角色（目前只有是否是志愿者）
      *