|
|
@ -2,13 +2,14 @@ package com.epmet.mq.listener; |
|
|
|
|
|
|
|
|
import com.alibaba.fastjson.JSON; |
|
|
import com.alibaba.fastjson.JSON; |
|
|
import com.epmet.commons.rocketmq.constants.MQUserPropertys; |
|
|
import com.epmet.commons.rocketmq.constants.MQUserPropertys; |
|
|
import com.epmet.commons.rocketmq.messages.CalPartyUnitSatisfactionFormDTO; |
|
|
import com.epmet.commons.rocketmq.messages.ServerSatisfactionCalFormDTO; |
|
|
import com.epmet.commons.tools.distributedlock.DistributedLock; |
|
|
import com.epmet.commons.tools.distributedlock.DistributedLock; |
|
|
import com.epmet.commons.tools.exception.ExceptionUtils; |
|
|
import com.epmet.commons.tools.exception.ExceptionUtils; |
|
|
import com.epmet.commons.tools.exception.RenException; |
|
|
import com.epmet.commons.tools.exception.RenException; |
|
|
import com.epmet.commons.tools.redis.RedisKeys; |
|
|
import com.epmet.commons.tools.redis.RedisKeys; |
|
|
import com.epmet.commons.tools.redis.RedisUtils; |
|
|
import com.epmet.commons.tools.redis.RedisUtils; |
|
|
import com.epmet.commons.tools.utils.SpringContextUtils; |
|
|
import com.epmet.commons.tools.utils.SpringContextUtils; |
|
|
|
|
|
import com.epmet.constant.SystemMessageType; |
|
|
import com.epmet.service.IcPartyUnitService; |
|
|
import com.epmet.service.IcPartyUnitService; |
|
|
import org.apache.commons.lang.StringUtils; |
|
|
import org.apache.commons.lang.StringUtils; |
|
|
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; |
|
|
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; |
|
|
@ -22,13 +23,7 @@ import org.slf4j.LoggerFactory; |
|
|
import java.util.List; |
|
|
import java.util.List; |
|
|
import java.util.concurrent.TimeUnit; |
|
|
import java.util.concurrent.TimeUnit; |
|
|
|
|
|
|
|
|
/** |
|
|
public class ServerSatisfactionCalEventListener implements MessageListenerConcurrently { |
|
|
* @Description 计算区域化党建单位，群众满意度=分数相加➗ 需求的总个数。 |
|
|
|
|
|
* @author wxz |
|
|
|
|
|
* @date 2021.10.13 15:21:48 |
|
|
|
|
|
*/ |
|
|
|
|
|
public class PartyUnitSatisfactionCalEventListener implements MessageListenerConcurrently { |
|
|
|
|
|
|
|
|
|
|
|
private Logger logger = LoggerFactory.getLogger(getClass()); |
|
|
private Logger logger = LoggerFactory.getLogger(getClass()); |
|
|
|
|
|
|
|
|
private RedisUtils redisUtils; |
|
|
private RedisUtils redisUtils; |
|
|
@ -57,23 +52,28 @@ public class PartyUnitSatisfactionCalEventListener implements MessageListenerCon |
|
|
String tags = messageExt.getTags(); |
|
|
String tags = messageExt.getTags(); |
|
|
String pendingMsgLabel = messageExt.getUserProperty(MQUserPropertys.BLOCKED_MSG_LABEL); |
|
|
String pendingMsgLabel = messageExt.getUserProperty(MQUserPropertys.BLOCKED_MSG_LABEL); |
|
|
|
|
|
|
|
|
logger.info("【计算区域化党建单位群众满意度事件监听器】-需求完成-收到消息内容:{},操作:{}", msg, tags); |
|
|
logger.info("【需求完成计算服务方群众满意度事件监听器】-需求完成-收到消息内容:{},操作:{}", msg, tags); |
|
|
CalPartyUnitSatisfactionFormDTO obj = JSON.parseObject(msg, CalPartyUnitSatisfactionFormDTO.class); |
|
|
ServerSatisfactionCalFormDTO obj = JSON.parseObject(msg, ServerSatisfactionCalFormDTO.class); |
|
|
|
|
|
|
|
|
DistributedLock distributedLock = null; |
|
|
DistributedLock distributedLock = null; |
|
|
RLock lock = null; |
|
|
RLock lock = null; |
|
|
try { |
|
|
try { |
|
|
distributedLock = SpringContextUtils.getBean(DistributedLock.class); |
|
|
distributedLock = SpringContextUtils.getBean(DistributedLock.class); |
|
|
lock = distributedLock.getLock(String.format("lock:ic_warn_stats:%s", obj.getCustomerId()), |
|
|
lock = distributedLock.getLock(String.format("lock:user_demand:%s", obj.getCustomerId()), |
|
|
30L, 30L, TimeUnit.SECONDS); |
|
|
30L, 30L, TimeUnit.SECONDS); |
|
|
//待执行方法
|
|
|
switch (tags) { |
|
|
SpringContextUtils.getBean(IcPartyUnitService.class).calPartyUnitSatisfation(obj); |
|
|
case SystemMessageType.CAL_PARTY_UNIT_SATISFACTION: |
|
|
|
|
|
//区域化党建单位-计算群众满意度
|
|
|
|
|
|
SpringContextUtils.getBean(IcPartyUnitService.class).calPartyUnitSatisfation(obj); |
|
|
|
|
|
break; |
|
|
|
|
|
//后面社会组织、社区自组织可能都会计算....先预备着
|
|
|
|
|
|
} |
|
|
} catch (RenException e) { |
|
|
} catch (RenException e) { |
|
|
// 如果是我们手动抛出的异常，说明在业务可控范围内。目前不需要MQ重试
|
|
|
// 如果是我们手动抛出的异常，说明在业务可控范围内。目前不需要MQ重试
|
|
|
logger.error("【计算区域化党建单位群众满意度事件监听器】-MQ失败:".concat(ExceptionUtils.getErrorStackTrace(e))); |
|
|
logger.error("【需求完成计算服务方群众满意度事件监听器】-MQ失败:".concat(ExceptionUtils.getErrorStackTrace(e))); |
|
|
} catch (Exception e) { |
|
|
} catch (Exception e) { |
|
|
// 不是我们自己抛出的异常，可以让MQ重试
|
|
|
// 不是我们自己抛出的异常，可以让MQ重试
|
|
|
logger.error("【计算区域化党建单位群众满意度监听器】-MQ失败:".concat(ExceptionUtils.getErrorStackTrace(e))); |
|
|
logger.error("【需求完成计算服务方群众满意度监听器】-MQ失败:".concat(ExceptionUtils.getErrorStackTrace(e))); |
|
|
throw e; |
|
|
throw e; |
|
|
} finally { |
|
|
} finally { |
|
|
distributedLock.unLock(lock); |
|
|
distributedLock.unLock(lock); |
|
|
@ -83,7 +83,7 @@ public class PartyUnitSatisfactionCalEventListener implements MessageListenerCon |
|
|
try { |
|
|
try { |
|
|
removePendingMqMsgCache(pendingMsgLabel); |
|
|
removePendingMqMsgCache(pendingMsgLabel); |
|
|
} catch (Exception e) { |
|
|
} catch (Exception e) { |
|
|
logger.error("【计算区域化党建单位群众满意度监听器】-删除mq阻塞消息缓存失败:{}", ExceptionUtils.getErrorStackTrace(e)); |
|
|
logger.error("【需求完成计算服务方群众满意度监听器】-删除mq阻塞消息缓存失败:{}", ExceptionUtils.getErrorStackTrace(e)); |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
yinzuomei
4 years ago