|  |  | @ -11,6 +11,7 @@ import com.epmet.commons.tools.redis.RedisUtils; | 
			
		
	
		
			
				
					|  |  |  | import com.epmet.commons.tools.utils.SpringContextUtils; | 
			
		
	
		
			
				
					|  |  |  | import com.epmet.constant.SystemMessageType; | 
			
		
	
		
			
				
					|  |  |  | import com.epmet.service.IcNatService; | 
			
		
	
		
			
				
					|  |  |  | import com.epmet.service.IcResiUserService; | 
			
		
	
		
			
				
					|  |  |  | import com.epmet.service.StatsResiWarnService; | 
			
		
	
		
			
				
					|  |  |  | import lombok.extern.slf4j.Slf4j; | 
			
		
	
		
			
				
					|  |  |  | import org.apache.commons.lang.StringUtils; | 
			
		
	
	
		
			
				
					|  |  | @ -26,10 +27,10 @@ import java.util.List; | 
			
		
	
		
			
				
					|  |  |  | import java.util.concurrent.TimeUnit; | 
			
		
	
		
			
				
					|  |  |  | 
 | 
			
		
	
		
			
				
					|  |  |  | /** | 
			
		
	
		
			
				
					|  |  |  |  * @Description 负能平台-客户居民信息变动监听器 | 
			
		
	
		
			
				
					|  |  |  |  * @author wxz | 
			
		
	
		
			
				
					|  |  |  |  * @Description 负能平台-居民信息变动监听器 | 
			
		
	
		
			
				
					|  |  |  |  * @date 2021.10.13 15:21:48 | 
			
		
	
		
			
				
					|  |  |  | */ | 
			
		
	
		
			
				
					|  |  |  |  */ | 
			
		
	
		
			
				
					|  |  |  | @Slf4j | 
			
		
	
		
			
				
					|  |  |  | public class ICWarnStatsEventListener implements MessageListenerConcurrently { | 
			
		
	
		
			
				
					|  |  |  | 
 | 
			
		
	
	
		
			
				
					|  |  | @ -37,6 +38,7 @@ public class ICWarnStatsEventListener implements MessageListenerConcurrently { | 
			
		
	
		
			
				
					|  |  |  | 
 | 
			
		
	
		
			
				
					|  |  |  |     private RedisUtils redisUtils; | 
			
		
	
		
			
				
					|  |  |  | 
 | 
			
		
	
		
			
				
					|  |  |  | 
 | 
			
		
	
		
			
				
					|  |  |  |     @Override | 
			
		
	
		
			
				
					|  |  |  |     public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { | 
			
		
	
		
			
				
					|  |  |  | 
 | 
			
		
	
	
		
			
				
					|  |  | @ -57,34 +59,25 @@ public class ICWarnStatsEventListener implements MessageListenerConcurrently { | 
			
		
	
		
			
				
					|  |  |  |         // msg即为消息体
 | 
			
		
	
		
			
				
					|  |  |  |         // tags为SystemMessageType.java中的项,为具体的操作,此处拿到tags,判断是创建还是变更,来做响应的后续操作即可
 | 
			
		
	
		
			
				
					|  |  |  |         String msg = new String(messageExt.getBody()); | 
			
		
	
		
			
				
					|  |  |  |         log.info("msg is {}",msg); | 
			
		
	
		
			
				
					|  |  |  |         log.info("msg is {}", msg); | 
			
		
	
		
			
				
					|  |  |  |         String topic = messageExt.getTopic(); | 
			
		
	
		
			
				
					|  |  |  |         String tags = messageExt.getTags(); | 
			
		
	
		
			
				
					|  |  |  |         String pendingMsgLabel = messageExt.getUserProperty(MQUserPropertys.BLOCKED_MSG_LABEL); | 
			
		
	
		
			
				
					|  |  |  | 
 | 
			
		
	
		
			
				
					|  |  |  |         logger.info("【开放数据事件监听器】-居民信息变动-收到消息内容:{},操作:{}", msg, tags); | 
			
		
	
		
			
				
					|  |  |  |         IcResiUserAddMQMsg obj = JSON.parseObject(msg, IcResiUserAddMQMsg.class); | 
			
		
	
		
			
				
					|  |  |  |         log.info("obj is {}",JSON.toJSONString(obj)); | 
			
		
	
		
			
				
					|  |  |  |         DistributedLock distributedLock = null; | 
			
		
	
		
			
				
					|  |  |  |         RLock lock = null; | 
			
		
	
		
			
				
					|  |  |  |         log.info("obj is {}", JSON.toJSONString(obj)); | 
			
		
	
		
			
				
					|  |  |  |         try { | 
			
		
	
		
			
				
					|  |  |  |             distributedLock = SpringContextUtils.getBean(DistributedLock.class); | 
			
		
	
		
			
				
					|  |  |  |             lock = distributedLock.getLock(String.format("lock:ic_warn_stats:%s", obj.getCustomerId()), | 
			
		
	
		
			
				
					|  |  |  |                     30L, 30L, TimeUnit.SECONDS); | 
			
		
	
		
			
				
					|  |  |  |             //待执行方法
 | 
			
		
	
		
			
				
					|  |  |  |             SpringContextUtils.getBean(StatsResiWarnService.class).resiWarn(obj.getCustomerId()); | 
			
		
	
		
			
				
					|  |  |  |             //根据居民Id更新 核酸检测的是否居民状态
 | 
			
		
	
		
			
				
					|  |  |  |             updateIcNatResiFlag(tags, obj); | 
			
		
	
		
			
				
					|  |  |  |             if (!resiWarnStats(obj)||!updateIcNatResiFlag(tags, obj)||!updateIcHouseResiNumber(obj)){ | 
			
		
	
		
			
				
					|  |  |  | 
 | 
			
		
	
		
			
				
					|  |  |  |             } | 
			
		
	
		
			
				
					|  |  |  |         } catch (RenException e) { | 
			
		
	
		
			
				
					|  |  |  |             // 如果是我们手动抛出的异常,说明在业务可控范围内。目前不需要MQ重试
 | 
			
		
	
		
			
				
					|  |  |  |             logger.error("【开放数据事件监听器】-客户居民信息变动MQ失败:".concat(ExceptionUtils.getErrorStackTrace(e))); | 
			
		
	
		
			
				
					|  |  |  |             logger.error("【开放数据事件监听器】-居民信息变动MQ失败:".concat(ExceptionUtils.getErrorStackTrace(e))); | 
			
		
	
		
			
				
					|  |  |  |         } catch (Exception e) { | 
			
		
	
		
			
				
					|  |  |  |             // 不是我们自己抛出的异常,可以让MQ重试
 | 
			
		
	
		
			
				
					|  |  |  |             logger.error("【开放数据事件监听器】-客户居民信息变动MQ失败:".concat(ExceptionUtils.getErrorStackTrace(e))); | 
			
		
	
		
			
				
					|  |  |  |             logger.error("【开放数据事件监听器】-居民信息变动MQ失败:".concat(ExceptionUtils.getErrorStackTrace(e))); | 
			
		
	
		
			
				
					|  |  |  |             throw e; | 
			
		
	
		
			
				
					|  |  |  |         } finally { | 
			
		
	
		
			
				
					|  |  |  |             distributedLock.unLock(lock); | 
			
		
	
		
			
				
					|  |  |  |         } | 
			
		
	
		
			
				
					|  |  |  | 
 | 
			
		
	
		
			
				
					|  |  |  |         if (StringUtils.isNotBlank(pendingMsgLabel)) { | 
			
		
	
	
		
			
				
					|  |  | @ -96,31 +89,97 @@ public class ICWarnStatsEventListener implements MessageListenerConcurrently { | 
			
		
	
		
			
				
					|  |  |  |         } | 
			
		
	
		
			
				
					|  |  |  |     } | 
			
		
	
		
			
				
					|  |  |  | 
 | 
			
		
	
		
			
				
					|  |  |  |     private boolean resiWarnStats(IcResiUserAddMQMsg obj) { | 
			
		
	
		
			
				
					|  |  |  |         DistributedLock distributedLock = null; | 
			
		
	
		
			
				
					|  |  |  |         RLock lock = null; | 
			
		
	
		
			
				
					|  |  |  |         try { | 
			
		
	
		
			
				
					|  |  |  |             if (StringUtils.isBlank(obj.getCustomerId())) { | 
			
		
	
		
			
				
					|  |  |  |                 log.warn("resiWarnStats param is error"); | 
			
		
	
		
			
				
					|  |  |  |                 return true; | 
			
		
	
		
			
				
					|  |  |  |             } | 
			
		
	
		
			
				
					|  |  |  |             distributedLock = SpringContextUtils.getBean(DistributedLock.class); | 
			
		
	
		
			
				
					|  |  |  |             lock = distributedLock.getLock(String.format("lock:ic_warn_stats:%s", obj.getCustomerId()), | 
			
		
	
		
			
				
					|  |  |  |                     30L, 30L, TimeUnit.SECONDS); | 
			
		
	
		
			
				
					|  |  |  |             SpringContextUtils.getBean(StatsResiWarnService.class).resiWarn(obj.getCustomerId()); | 
			
		
	
		
			
				
					|  |  |  |             log.debug("【居民信息变动】resiWarnStats param customerId:{}", JSON.toJSONString(obj)); | 
			
		
	
		
			
				
					|  |  |  | 
 | 
			
		
	
		
			
				
					|  |  |  |         } catch (Exception e) { | 
			
		
	
		
			
				
					|  |  |  |             log.error("resiWarnStats exception", e); | 
			
		
	
		
			
				
					|  |  |  |             return false; | 
			
		
	
		
			
				
					|  |  |  |         } finally { | 
			
		
	
		
			
				
					|  |  |  |             if (distributedLock != null){ | 
			
		
	
		
			
				
					|  |  |  |                 distributedLock.unLock(lock); | 
			
		
	
		
			
				
					|  |  |  |             } | 
			
		
	
		
			
				
					|  |  |  |         } | 
			
		
	
		
			
				
					|  |  |  |         return true; | 
			
		
	
		
			
				
					|  |  |  |     } | 
			
		
	
		
			
				
					|  |  |  | 
 | 
			
		
	
		
			
				
					|  |  |  |     /** | 
			
		
	
		
			
				
					|  |  |  |      * desc:根据客户id更新 每个房屋内居民的总数 | 
			
		
	
		
			
				
					|  |  |  |      * @param obj | 
			
		
	
		
			
				
					|  |  |  |      * @return | 
			
		
	
		
			
				
					|  |  |  |      * @remark 因为使用了线程 所以结果不用关心 | 
			
		
	
		
			
				
					|  |  |  |      */ | 
			
		
	
		
			
				
					|  |  |  |     private boolean updateIcHouseResiNumber(IcResiUserAddMQMsg obj) { | 
			
		
	
		
			
				
					|  |  |  |         DistributedLock distributedLock = null; | 
			
		
	
		
			
				
					|  |  |  |         RLock lock = null; | 
			
		
	
		
			
				
					|  |  |  |         try { | 
			
		
	
		
			
				
					|  |  |  |             if (StringUtils.isBlank(obj.getCustomerId())) { | 
			
		
	
		
			
				
					|  |  |  |                 log.warn("updateIcHouseResiNumber param is error"); | 
			
		
	
		
			
				
					|  |  |  |                 return true; | 
			
		
	
		
			
				
					|  |  |  |             } | 
			
		
	
		
			
				
					|  |  |  |             distributedLock = SpringContextUtils.getBean(DistributedLock.class); | 
			
		
	
		
			
				
					|  |  |  |             lock = distributedLock.getLock(String.format("lock:update_house_resi_number:%s", obj.getCustomerId()), | 
			
		
	
		
			
				
					|  |  |  |                     30L, 30L, TimeUnit.SECONDS); | 
			
		
	
		
			
				
					|  |  |  |             SpringContextUtils.getBean(IcResiUserService.class).updateIcHouseResiNumber(obj.getCustomerId()); | 
			
		
	
		
			
				
					|  |  |  |             log.debug("【居民信息变动】updateIcHouseResiNumber param customerId & icResiUserId:{}", JSON.toJSONString(obj)); | 
			
		
	
		
			
				
					|  |  |  |         } catch (Exception e) { | 
			
		
	
		
			
				
					|  |  |  |             log.error("updateIcHouseResiNumber exception", e); | 
			
		
	
		
			
				
					|  |  |  |             return false; | 
			
		
	
		
			
				
					|  |  |  |         } finally { | 
			
		
	
		
			
				
					|  |  |  |             if (distributedLock != null){ | 
			
		
	
		
			
				
					|  |  |  |                 distributedLock.unLock(lock); | 
			
		
	
		
			
				
					|  |  |  |             } | 
			
		
	
		
			
				
					|  |  |  |         } | 
			
		
	
		
			
				
					|  |  |  |         return true; | 
			
		
	
		
			
				
					|  |  |  |     } | 
			
		
	
		
			
				
					|  |  |  | 
 | 
			
		
	
		
			
				
					|  |  |  |     /** | 
			
		
	
		
			
				
					|  |  |  |      * desc:根据居民Id更新 核酸检测的是否居民状态(只有新增和删除时调用) | 
			
		
	
		
			
				
					|  |  |  |      * @param tags | 
			
		
	
		
			
				
					|  |  |  |      *  @param tags | 
			
		
	
		
			
				
					|  |  |  |      * @param obj | 
			
		
	
		
			
				
					|  |  |  |      * @return | 
			
		
	
		
			
				
					|  |  |  |      */ | 
			
		
	
		
			
				
					|  |  |  |     private void updateIcNatResiFlag(String tags, IcResiUserAddMQMsg obj) { | 
			
		
	
		
			
				
					|  |  |  |     private boolean updateIcNatResiFlag(String tags, IcResiUserAddMQMsg obj) { | 
			
		
	
		
			
				
					|  |  |  |         DistributedLock distributedLock = null; | 
			
		
	
		
			
				
					|  |  |  |         RLock lock = null; | 
			
		
	
		
			
				
					|  |  |  |         try { | 
			
		
	
		
			
				
					|  |  |  |             if (StringUtils.isBlank(obj.getCustomerId())||StringUtils.isBlank(obj.getIcResiUser())){ | 
			
		
	
		
			
				
					|  |  |  |             if (StringUtils.isBlank(obj.getCustomerId()) || StringUtils.isBlank(obj.getIcResiUser())) { | 
			
		
	
		
			
				
					|  |  |  |                 log.warn("updateIcNatResiFlag param is error"); | 
			
		
	
		
			
				
					|  |  |  |                 return; | 
			
		
	
		
			
				
					|  |  |  |                 return true; | 
			
		
	
		
			
				
					|  |  |  |             } | 
			
		
	
		
			
				
					|  |  |  |             distributedLock = SpringContextUtils.getBean(DistributedLock.class); | 
			
		
	
		
			
				
					|  |  |  |             lock = distributedLock.getLock(String.format("lock:ic_nat_resi_flag:%s", obj.getCustomerId()), | 
			
		
	
		
			
				
					|  |  |  |                     30L, 30L, TimeUnit.SECONDS); | 
			
		
	
		
			
				
					|  |  |  |             if (SystemMessageType.IC_RESI_USER_ADD.equals(tags) || SystemMessageType.IC_RESI_USER_DEL.equals(tags)) { | 
			
		
	
		
			
				
					|  |  |  |                 Integer effectRow = SpringContextUtils.getBean(IcNatService.class).updateIsResiFlag(obj.getCustomerId(), obj.getIcResiUser()); | 
			
		
	
		
			
				
					|  |  |  |                 log.debug("updateIcNatResiFlag effectRow:{},param customerId & icResiUserId:{}", effectRow, JSON.toJSONString(obj)); | 
			
		
	
		
			
				
					|  |  |  |                 log.debug("【居民信息变动】updateIcNatResiFlag effectRow:{},param customerId & icResiUserId:{}", effectRow, JSON.toJSONString(obj)); | 
			
		
	
		
			
				
					|  |  |  |             } | 
			
		
	
		
			
				
					|  |  |  |         } catch (Exception e) { | 
			
		
	
		
			
				
					|  |  |  |             log.error("updateIcNatResiFlag exception", e); | 
			
		
	
		
			
				
					|  |  |  |             return false; | 
			
		
	
		
			
				
					|  |  |  |         } finally { | 
			
		
	
		
			
				
					|  |  |  |             if (distributedLock != null){ | 
			
		
	
		
			
				
					|  |  |  |                 distributedLock.unLock(lock); | 
			
		
	
		
			
				
					|  |  |  |             } | 
			
		
	
		
			
				
					|  |  |  |         } | 
			
		
	
		
			
				
					|  |  |  |         return true; | 
			
		
	
		
			
				
					|  |  |  |     } | 
			
		
	
		
			
				
					|  |  |  | 
 | 
			
		
	
		
			
				
					|  |  |  |     /** | 
			
		
	
		
			
				
					|  |  |  |      * @description | 
			
		
	
		
			
				
					|  |  |  |      * | 
			
		
	
		
			
				
					|  |  |  |      * @param pendingMsgLabel | 
			
		
	
		
			
				
					|  |  |  |      * @return | 
			
		
	
		
			
				
					|  |  |  |      * @description | 
			
		
	
		
			
				
					|  |  |  |      * @author wxz | 
			
		
	
		
			
				
					|  |  |  |      * @date 2021.10.14 16:32:32 | 
			
		
	
		
			
				
					|  |  |  |      */ | 
			
		
	
	
		
			
				
					|  |  | 
 |