yinzuomei
4 years ago
9 changed files with 220 additions and 1 deletions
@ -0,0 +1,31 @@ |
|||
package com.epmet.mq; |
|||
|
|||
import com.epmet.commons.rocketmq.constants.ConsomerGroupConstants; |
|||
import com.epmet.commons.rocketmq.constants.TopicConstants; |
|||
import com.epmet.commons.rocketmq.register.MQAbstractRegister; |
|||
import com.epmet.commons.rocketmq.register.MQConsumerProperties; |
|||
import com.epmet.mq.listener.PartyUnitSatisfactionCalEventListener; |
|||
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel; |
|||
import org.springframework.stereotype.Component; |
|||
|
|||
/** |
|||
* @Description 如果rocketmq.enable=true，这里必须实现，且 实例化 |
|||
* @author wxz |
|||
* @date 2021.07.14 17:13:41 |
|||
*/ |
|||
@Component |
|||
public class RocketMQConsumerRegister extends MQAbstractRegister { |
|||
|
|||
@Override |
|||
public void registerAllListeners(String env, MQConsumerProperties consumerProperties) { |
|||
// 客户初始化监听器注册
|
|||
register(consumerProperties, |
|||
ConsomerGroupConstants.CAL_PARTY_UNIT_SATISFACTION, |
|||
MessageModel.CLUSTERING, |
|||
TopicConstants.CAL_PARTY_UNIT_SATISFACTION, |
|||
"*", |
|||
new PartyUnitSatisfactionCalEventListener()); |
|||
|
|||
// ...其他监听器类似
|
|||
} |
|||
} |
|||
@ -0,0 +1,104 @@ |
|||
package com.epmet.mq.listener; |
|||
|
|||
import com.alibaba.fastjson.JSON; |
|||
import com.epmet.commons.rocketmq.constants.MQUserPropertys; |
|||
import com.epmet.commons.rocketmq.messages.CalPartyUnitSatisfactionFormDTO; |
|||
import com.epmet.commons.tools.distributedlock.DistributedLock; |
|||
import com.epmet.commons.tools.exception.ExceptionUtils; |
|||
import com.epmet.commons.tools.exception.RenException; |
|||
import com.epmet.commons.tools.redis.RedisKeys; |
|||
import com.epmet.commons.tools.redis.RedisUtils; |
|||
import com.epmet.commons.tools.utils.SpringContextUtils; |
|||
import com.epmet.service.IcPartyUnitService; |
|||
import org.apache.commons.lang.StringUtils; |
|||
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; |
|||
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; |
|||
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; |
|||
import org.apache.rocketmq.common.message.MessageExt; |
|||
import org.redisson.api.RLock; |
|||
import org.slf4j.Logger; |
|||
import org.slf4j.LoggerFactory; |
|||
|
|||
import java.util.List; |
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
/** |
|||
* @Description 计算区域化党建单位，群众满意度=分数相加➗ 需求的总个数。 |
|||
* @author wxz |
|||
* @date 2021.10.13 15:21:48 |
|||
*/ |
|||
public class PartyUnitSatisfactionCalEventListener implements MessageListenerConcurrently { |
|||
|
|||
private Logger logger = LoggerFactory.getLogger(getClass()); |
|||
|
|||
private RedisUtils redisUtils; |
|||
|
|||
@Override |
|||
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { |
|||
|
|||
if (redisUtils == null) { |
|||
redisUtils = SpringContextUtils.getBean(RedisUtils.class); |
|||
} |
|||
|
|||
try { |
|||
msgs.forEach(msg -> consumeMessage(msg)); |
|||
} catch (Exception e) { |
|||
logger.error(ExceptionUtils.getErrorStackTrace(e)); |
|||
return ConsumeConcurrentlyStatus.RECONSUME_LATER; |
|||
} |
|||
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; |
|||
} |
|||
|
|||
private void consumeMessage(MessageExt messageExt) { |
|||
// msg即为消息体
|
|||
// tags为SystemMessageType.java中的项，为具体的操作，此处拿到tags，判断是创建还是变更，来做响应的后续操作即可
|
|||
String msg = new String(messageExt.getBody()); |
|||
String topic = messageExt.getTopic(); |
|||
String tags = messageExt.getTags(); |
|||
String pendingMsgLabel = messageExt.getUserProperty(MQUserPropertys.BLOCKED_MSG_LABEL); |
|||
|
|||
logger.info("【计算区域化党建单位群众满意度事件监听器】-需求完成-收到消息内容:{},操作:{}", msg, tags); |
|||
CalPartyUnitSatisfactionFormDTO obj = JSON.parseObject(msg, CalPartyUnitSatisfactionFormDTO.class); |
|||
|
|||
DistributedLock distributedLock = null; |
|||
RLock lock = null; |
|||
try { |
|||
distributedLock = SpringContextUtils.getBean(DistributedLock.class); |
|||
lock = distributedLock.getLock(String.format("lock:ic_warn_stats:%s", obj.getCustomerId()), |
|||
30L, 30L, TimeUnit.SECONDS); |
|||
//待执行方法
|
|||
SpringContextUtils.getBean(IcPartyUnitService.class).calPartyUnitSatisfation(obj); |
|||
} catch (RenException e) { |
|||
// 如果是我们手动抛出的异常，说明在业务可控范围内。目前不需要MQ重试
|
|||
logger.error("【计算区域化党建单位群众满意度事件监听器】-MQ失败:".concat(ExceptionUtils.getErrorStackTrace(e))); |
|||
} catch (Exception e) { |
|||
// 不是我们自己抛出的异常，可以让MQ重试
|
|||
logger.error("【计算区域化党建单位群众满意度监听器】-MQ失败:".concat(ExceptionUtils.getErrorStackTrace(e))); |
|||
throw e; |
|||
} finally { |
|||
distributedLock.unLock(lock); |
|||
} |
|||
|
|||
if (StringUtils.isNotBlank(pendingMsgLabel)) { |
|||
try { |
|||
removePendingMqMsgCache(pendingMsgLabel); |
|||
} catch (Exception e) { |
|||
logger.error("【计算区域化党建单位群众满意度监听器】-删除mq阻塞消息缓存失败:{}", ExceptionUtils.getErrorStackTrace(e)); |
|||
} |
|||
} |
|||
} |
|||
|
|||
/** |
|||
* @description |
|||
* |
|||
* @param pendingMsgLabel |
|||
* @return |
|||
* @author wxz |
|||
* @date 2021.10.14 16:32:32 |
|||
*/ |
|||
private void removePendingMqMsgCache(String pendingMsgLabel) { |
|||
String key = RedisKeys.blockedMqMsgKey(pendingMsgLabel); |
|||
redisUtils.delete(key); |
|||
//logger.info("【开放数据事件监听器】删除mq阻塞消息缓存成功,blockedMsgLabel:{}", pendingMsgLabel);
|
|||
} |
|||
} |
|||
Loading…
Reference in new issue