wangxianzhang
4 years ago
17 changed files with 566 additions and 84 deletions
@ -0,0 +1,27 @@ |
|||||
|
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.PointListener; |
||||
|
import com.epmet.service.UserPointActionLogService; |
||||
|
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel; |
||||
|
import org.springframework.beans.factory.annotation.Autowired; |
||||
|
import org.springframework.stereotype.Component; |
||||
|
|
||||
|
@Component |
||||
|
public class RocketMQConsumerRegister extends MQAbstractRegister { |
||||
|
|
||||
|
@Autowired |
||||
|
private UserPointActionLogService userPointActionLogService; |
||||
|
|
||||
|
@Override |
||||
|
public void registerAllListeners(String env, MQConsumerProperties consumerProperties) { |
||||
|
// 客户初始化监听器注册
|
||||
|
register(consumerProperties, ConsomerGroupConstants.RESI_GROUP_POINT_OPERATION_GROUP, MessageModel.CLUSTERING, TopicConstants.RESI_GROUP, "*", new PointListener(userPointActionLogService)); |
||||
|
register(consumerProperties, ConsomerGroupConstants.EPMET_HEART_POINT_OPERATION_GROUP, MessageModel.CLUSTERING, TopicConstants.EPMET_HEART, "*", new PointListener(userPointActionLogService)); |
||||
|
|
||||
|
// ...其他监听器类似
|
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,236 @@ |
|||||
|
package com.epmet.mq.listener; |
||||
|
|
||||
|
import com.alibaba.fastjson.JSON; |
||||
|
import com.epmet.commons.rocketmq.constants.MQUserPropertys; |
||||
|
import com.epmet.commons.rocketmq.messages.LoginMQMsg; |
||||
|
import com.epmet.commons.tools.distributedlock.DistributedLock; |
||||
|
import com.epmet.commons.tools.dto.form.mq.eventmsg.BasePointEventMsg; |
||||
|
import com.epmet.commons.tools.enums.EventEnum; |
||||
|
import com.epmet.commons.tools.exception.EpmetErrorCode; |
||||
|
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.Result; |
||||
|
import com.epmet.commons.tools.utils.SpringContextUtils; |
||||
|
import com.epmet.constant.SystemMessageType; |
||||
|
import com.epmet.service.UserPointActionLogService; |
||||
|
import dto.form.SendPointFormDTO; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.apache.commons.lang3.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; |
||||
|
|
||||
|
/** |
||||
|
* @author wxz |
||||
|
* @Description 登录操作日志监听器 |
||||
|
|
||||
|
* @return |
||||
|
* @date 2021.06.07 16:12 |
||||
|
*/ |
||||
|
@Slf4j |
||||
|
public class PointListener implements MessageListenerConcurrently { |
||||
|
|
||||
|
private RedisUtils redisUtils; |
||||
|
|
||||
|
private UserPointActionLogService userPointActionLogService; |
||||
|
|
||||
|
public PointListener(UserPointActionLogService userPointActionLogService) { |
||||
|
this.userPointActionLogService = userPointActionLogService; |
||||
|
} |
||||
|
|
||||
|
@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) { |
||||
|
log.error(ExceptionUtils.getErrorStackTrace(e)); |
||||
|
return ConsumeConcurrentlyStatus.RECONSUME_LATER; |
||||
|
} |
||||
|
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; |
||||
|
} |
||||
|
|
||||
|
private void consumeMessage(MessageExt messageExt) { |
||||
|
String tags = messageExt.getTags(); |
||||
|
String msg = new String(messageExt.getBody()); |
||||
|
String pendingMsgLabel = messageExt.getUserProperty(MQUserPropertys.BLOCKED_MSG_LABEL); |
||||
|
log.info("【积分操作监听器】-收到消息内容:{}", msg); |
||||
|
|
||||
|
DistributedLock distributedLock = null; |
||||
|
RLock lock = null; |
||||
|
try { |
||||
|
distributedLock = SpringContextUtils.getBean(DistributedLock.class); |
||||
|
lock = distributedLock.getLock("lock:point_operation", 30L, 30L, TimeUnit.SECONDS); |
||||
|
|
||||
|
switch (tags) { |
||||
|
case SystemMessageType.PARTICIPATE_ONE_TOPIC: |
||||
|
resiGroupPointOpe(msg); |
||||
|
break; |
||||
|
case SystemMessageType.ACTIVE_SEND_POINT: |
||||
|
// 爱心活动积分发放
|
||||
|
activeSendPoint(msg); |
||||
|
break; |
||||
|
case SystemMessageType.ACTIVE_INSERT_LIVE: |
||||
|
// 添加活动实况
|
||||
|
activeInsertLive(msg); |
||||
|
break; |
||||
|
case SystemMessageType.REGISTER_VOLUNTEER: |
||||
|
// 注册志愿者
|
||||
|
registerVolunteer(msg); |
||||
|
break; |
||||
|
case SystemMessageType.TOPIC_TO_ISSUE: |
||||
|
// 话题被组长转为议题
|
||||
|
topicToIssue(msg); |
||||
|
break; |
||||
|
case SystemMessageType.TOPIC_TO_PROJECT: |
||||
|
// 话题转为项目
|
||||
|
topicToProject(msg); |
||||
|
break; |
||||
|
case SystemMessageType.INVITE_RESI_INTO_GROUP: |
||||
|
inviteResiIntoGroup(msg); |
||||
|
break; |
||||
|
case SystemMessageType.INVITE_NEW_INTO_GROUP: |
||||
|
inviteNewIntoGroup(msg); |
||||
|
break; |
||||
|
case SystemMessageType.SHIFT_TOPIC_TO_ISSUE: |
||||
|
shiftTopicToIssue(msg); |
||||
|
break; |
||||
|
case SystemMessageType.PUBLISH_ONE_TOPIC: |
||||
|
publicOneTopic(msg); |
||||
|
break; |
||||
|
case SystemMessageType.LEADER_RESOLVE_TOPIC: |
||||
|
leaderResolveTopic(msg); |
||||
|
break; |
||||
|
} |
||||
|
|
||||
|
} catch (RenException e) { |
||||
|
// 如果是我们手动抛出的异常，说明在业务可控范围内。目前不需要MQ重试
|
||||
|
log.error("【积分操作监听器】消费消息失败:".concat(ExceptionUtils.getErrorStackTrace(e))); |
||||
|
} catch (Exception e) { |
||||
|
// 不是我们自己抛出的异常，可以让MQ重试
|
||||
|
log.error("【积分操作监听器】消费消息失败:".concat(ExceptionUtils.getErrorStackTrace(e))); |
||||
|
throw e; |
||||
|
} finally { |
||||
|
distributedLock.unLock(lock); |
||||
|
} |
||||
|
|
||||
|
if (StringUtils.isNotBlank(pendingMsgLabel)) { |
||||
|
try { |
||||
|
removePendingMqMsgCache(pendingMsgLabel); |
||||
|
} catch (Exception e) { |
||||
|
log.error("【登录操作事件监听器】-删除mq阻塞消息缓存失败:{}", ExceptionUtils.getErrorStackTrace(e)); |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
private void leaderResolveTopic(String msg) { |
||||
|
List<BasePointEventMsg> formList = JSON.parseArray(msg, BasePointEventMsg.class); |
||||
|
|
||||
|
formList.forEach(obj -> { |
||||
|
userPointActionLogService.grantPointByEvent(obj.getEventTag(),obj); |
||||
|
}); |
||||
|
} |
||||
|
|
||||
|
private void publicOneTopic(String msg) { |
||||
|
List<BasePointEventMsg> formList = JSON.parseArray(msg, BasePointEventMsg.class); |
||||
|
|
||||
|
formList.forEach(obj -> { |
||||
|
userPointActionLogService.grantPointByEvent(obj.getEventTag(),obj); |
||||
|
}); |
||||
|
} |
||||
|
|
||||
|
private void shiftTopicToIssue(String msg) { |
||||
|
List<BasePointEventMsg> formList = JSON.parseArray(msg, BasePointEventMsg.class); |
||||
|
|
||||
|
formList.forEach(obj -> { |
||||
|
userPointActionLogService.grantPointByEvent(obj.getEventTag(),obj); |
||||
|
}); |
||||
|
} |
||||
|
|
||||
|
private void inviteNewIntoGroup(String msg) { |
||||
|
List<BasePointEventMsg> formList = JSON.parseArray(msg, BasePointEventMsg.class); |
||||
|
|
||||
|
formList.forEach(obj -> { |
||||
|
userPointActionLogService.grantPointByEvent(obj.getEventTag(),obj); |
||||
|
}); |
||||
|
} |
||||
|
|
||||
|
private void inviteResiIntoGroup(String msg) { |
||||
|
List<BasePointEventMsg> formList = JSON.parseArray(msg, BasePointEventMsg.class); |
||||
|
|
||||
|
formList.forEach(obj -> { |
||||
|
userPointActionLogService.grantPointByEvent(obj.getEventTag(),obj); |
||||
|
}); |
||||
|
} |
||||
|
|
||||
|
private void topicToProject(String msg) { |
||||
|
List<BasePointEventMsg> formList = JSON.parseArray(msg, BasePointEventMsg.class); |
||||
|
|
||||
|
formList.forEach(obj -> { |
||||
|
userPointActionLogService.grantPointByEvent(obj.getEventTag(),obj); |
||||
|
}); |
||||
|
} |
||||
|
|
||||
|
private void topicToIssue(String msg) { |
||||
|
List<BasePointEventMsg> formList = JSON.parseArray(msg, BasePointEventMsg.class); |
||||
|
|
||||
|
formList.forEach(obj -> { |
||||
|
userPointActionLogService.grantPointByEvent(obj.getEventTag(), obj); |
||||
|
}); |
||||
|
} |
||||
|
|
||||
|
private void registerVolunteer(String msg) { |
||||
|
List<BasePointEventMsg> formList = JSON.parseArray(msg, BasePointEventMsg.class); |
||||
|
formList.forEach(obj -> { |
||||
|
userPointActionLogService.grantPointByEvent(EventEnum.REGISTER_VOLUNTEER.getEventTag(),obj); |
||||
|
}); |
||||
|
} |
||||
|
|
||||
|
private void activeInsertLive(String msg) { |
||||
|
List<BasePointEventMsg> formList = JSON.parseArray(msg, BasePointEventMsg.class); |
||||
|
formList.forEach(obj -> { |
||||
|
userPointActionLogService.grantPointByEvent(EventEnum.ACTIVE_INSERT_LIVE.getEventTag(), obj); |
||||
|
}); |
||||
|
} |
||||
|
|
||||
|
private void activeSendPoint(String msg) { |
||||
|
List<SendPointFormDTO> formDTO = JSON.parseArray(msg, SendPointFormDTO.class); |
||||
|
userPointActionLogService.grantPoint(formDTO); |
||||
|
} |
||||
|
|
||||
|
private void resiGroupPointOpe(String msg) { |
||||
|
List<BasePointEventMsg> formList = JSON.parseArray(msg, BasePointEventMsg.class); |
||||
|
|
||||
|
formList.forEach(obj -> { |
||||
|
userPointActionLogService.grantPointByEvent(obj.getEventTag(),obj); |
||||
|
}); |
||||
|
} |
||||
|
|
||||
|
|
||||
|
/** |
||||
|
* @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("【登录操作事件监听器】删除pendingMsgLabel成功:{}", pendingMsgLabel);
|
||||
|
} |
||||
|
} |
||||
Loading…
Reference in new issue