sunyuchao
4 years ago
11 changed files with 217 additions and 46 deletions
@ -0,0 +1,20 @@ |
|||||
|
package com.epmet.commons.rocketmq.messages; |
||||
|
|
||||
|
import lombok.Data; |
||||
|
|
||||
|
import java.io.Serializable; |
||||
|
|
||||
|
/** |
||||
|
* 居民信息新增、修改推送MQ |
||||
|
* @author sun |
||||
|
*/ |
||||
|
@Data |
||||
|
public class IcWarnStatsMQMsg implements Serializable { |
||||
|
|
||||
|
//客户Id
|
||||
|
private String customerId; |
||||
|
//居民ID
|
||||
|
private String icResiUser; |
||||
|
|
||||
|
|
||||
|
} |
||||
@ -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.ICWarnStatsEventListener; |
||||
|
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 RocketMQWarnStatsRegister extends MQAbstractRegister { |
||||
|
|
||||
|
@Override |
||||
|
public void registerAllListeners(String env, MQConsumerProperties consumerProperties) { |
||||
|
// 客户初始化监听器注册
|
||||
|
register(consumerProperties, |
||||
|
ConsomerGroupConstants.IC_WARN_STATS_EVENT_LISTENER_GROUP, |
||||
|
MessageModel.CLUSTERING, |
||||
|
TopicConstants.IC_RESI_USER, |
||||
|
"*", |
||||
|
new ICWarnStatsEventListener()); |
||||
|
|
||||
|
// ...其他监听器类似
|
||||
|
} |
||||
|
} |
||||
@ -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.IcWarnStatsMQMsg; |
||||
|
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 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 ICWarnStatsEventListener 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); |
||||
|
IcWarnStatsMQMsg obj = JSON.parseObject(msg, IcWarnStatsMQMsg.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); |
||||
|
System.out.println("嘻嘻哈哈哈乐乐呵呵-----------"); |
||||
|
//待执行方法
|
||||
|
//SpringContextUtils.getBean(BaseGridInfoService.class).getAgencyBaseInfo(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);
|
||||
|
} |
||||
|
} |
||||
@ -1,39 +0,0 @@ |
|||||
//package com.epmet.mq.listener;
|
|
||||
//
|
|
||||
//import com.alibaba.fastjson.JSON;
|
|
||||
//import com.epmet.commons.rocketmq.constants.ConsomerGroupConstants;
|
|
||||
//import com.epmet.commons.rocketmq.constants.TopicConstants;
|
|
||||
//import com.epmet.commons.rocketmq.messages.InitCustomerMQMsg;
|
|
||||
//import com.epmet.service.GovStaffRoleService;
|
|
||||
//import org.apache.rocketmq.common.message.MessageExt;
|
|
||||
//import org.apache.rocketmq.spring.annotation.MessageModel;
|
|
||||
//import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
|
|
||||
//import org.apache.rocketmq.spring.core.RocketMQListener;
|
|
||||
//import org.slf4j.Logger;
|
|
||||
//import org.slf4j.LoggerFactory;
|
|
||||
//import org.springframework.beans.factory.annotation.Autowired;
|
|
||||
//import org.springframework.stereotype.Component;
|
|
||||
//
|
|
||||
///**
|
|
||||
// * 监听初始化客户动作，为客户初始化角色列表
|
|
||||
// */
|
|
||||
//@RocketMQMessageListener(topic = TopicConstants.INIT_CUSTOMER,
|
|
||||
// consumerGroup = ConsomerGroupConstants.INIT_CUSTOMER_ROLES_GROUP,
|
|
||||
// messageModel = MessageModel.CLUSTERING,
|
|
||||
// selectorExpression = "*")
|
|
||||
//@Component
|
|
||||
//public class InitCustomerRolesListener implements RocketMQListener<MessageExt> {
|
|
||||
//
|
|
||||
// private Logger logger = LoggerFactory.getLogger(getClass());
|
|
||||
//
|
|
||||
// @Autowired
|
|
||||
// private GovStaffRoleService govStaffRoleService;
|
|
||||
//
|
|
||||
// @Override
|
|
||||
// public void onMessage(MessageExt messageExt) {
|
|
||||
// String msg = new String(messageExt.getBody());
|
|
||||
// logger.info("初始化客户-初始化角色列表-收到消息内容:{}", msg);
|
|
||||
// InitCustomerMQMsg msgObj = JSON.parseObject(msg, InitCustomerMQMsg.class);
|
|
||||
// govStaffRoleService.initGovStaffRolesForCustomer(msgObj.getCustomerId());
|
|
||||
// }
|
|
||||
//}
|
|
||||
Loading…
Reference in new issue