wxz
4 years ago
11 changed files with 237 additions and 11 deletions
@ -0,0 +1,91 @@ |
|||||
|
package com.epmet.mq.listener.listener; |
||||
|
|
||||
|
import com.alibaba.fastjson.JSON; |
||||
|
import com.epmet.commons.rocketmq.messages.PointRuleChangedMQMsg; |
||||
|
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.utils.SpringContextUtils; |
||||
|
import com.epmet.entity.LogOperationEntity; |
||||
|
import com.epmet.enums.SystemMessageTypeEnum; |
||||
|
import com.epmet.mq.listener.bean.log.LogOperationHelper; |
||||
|
import com.epmet.mq.listener.bean.log.OperatorInfo; |
||||
|
import com.epmet.service.LogOperationService; |
||||
|
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.21 10:13 |
||||
|
*/ |
||||
|
public class PointOperationLogListener implements MessageListenerConcurrently { |
||||
|
|
||||
|
private Logger logger = LoggerFactory.getLogger(getClass()); |
||||
|
|
||||
|
@Override |
||||
|
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { |
||||
|
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) { |
||||
|
String opeType = messageExt.getTags(); |
||||
|
String msg = new String(messageExt.getBody()); |
||||
|
logger.info("积分操作日志监听器-收到消息内容:{}", msg); |
||||
|
PointRuleChangedMQMsg msgObj = JSON.parseObject(msg, PointRuleChangedMQMsg.class); |
||||
|
|
||||
|
String content = StringUtils.isBlank(msgObj.getOperationBrief()) ? "" : msgObj.getOperationBrief(); |
||||
|
|
||||
|
OperatorInfo operatorInfo = LogOperationHelper.getInstance().getOperatorInfo(msgObj.getOperatorId()); |
||||
|
|
||||
|
LogOperationEntity logEntity = new LogOperationEntity(); |
||||
|
logEntity.setCategory(messageExt.getTopic()); |
||||
|
logEntity.setType(opeType); |
||||
|
logEntity.setTypeDisplay(SystemMessageTypeEnum.getTypeDisplay(opeType)); |
||||
|
logEntity.setTargetId(msgObj.getRuleId()); |
||||
|
logEntity.setIp(msgObj.getIp()); |
||||
|
logEntity.setFromApp(msgObj.getFromApp()); |
||||
|
logEntity.setFromClient(msgObj.getFromClient()); |
||||
|
logEntity.setCustomerId(operatorInfo.getCustomerId()); |
||||
|
logEntity.setOperatorId(msgObj.getOperatorId()); |
||||
|
logEntity.setOperatorMobile(operatorInfo.getMobile()); |
||||
|
logEntity.setOperatorName(operatorInfo.getName()); |
||||
|
logEntity.setOperatingTime(msgObj.getOperatingTime()); |
||||
|
logEntity.setContent(content); |
||||
|
|
||||
|
DistributedLock distributedLock = null; |
||||
|
RLock lock = null; |
||||
|
try { |
||||
|
distributedLock = SpringContextUtils.getBean(DistributedLock.class); |
||||
|
lock = distributedLock.getLock(String.format("lock:point_operation_log:%s:%s", logEntity.getType(), logEntity.getTargetId()), |
||||
|
30L, 30L, TimeUnit.SECONDS); |
||||
|
SpringContextUtils.getBean(LogOperationService.class).log(logEntity); |
||||
|
} catch (RenException e) { |
||||
|
// 如果是我们手动抛出的异常，说明在业务可控范围内。目前不需要MQ重试
|
||||
|
logger.error("【RocketMQ】添加操作日志失败:".concat(ExceptionUtils.getErrorStackTrace(e))); |
||||
|
} catch (Exception e) { |
||||
|
// 不是我们自己抛出的异常，可以让MQ重试
|
||||
|
logger.error("【RocketMQ】添加操作日志失败:".concat(ExceptionUtils.getErrorStackTrace(e))); |
||||
|
throw e; |
||||
|
} finally { |
||||
|
distributedLock.unLock(lock); |
||||
|
} |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,33 @@ |
|||||
|
package com.epmet.commons.rocketmq.messages; |
||||
|
|
||||
|
import lombok.Data; |
||||
|
|
||||
|
import java.util.Date; |
||||
|
|
||||
|
/** |
||||
|
* 积分规则变动消息体 |
||||
|
*/ |
||||
|
@Data |
||||
|
public class PointRuleChangedMQMsg { |
||||
|
/** |
||||
|
* 操作简介 |
||||
|
*/ |
||||
|
private String operationBrief; |
||||
|
|
||||
|
/** |
||||
|
* 规则的id |
||||
|
*/ |
||||
|
private String ruleId; |
||||
|
/** |
||||
|
* 谁操作的 |
||||
|
*/ |
||||
|
private String operatorId; |
||||
|
|
||||
|
private String ip; |
||||
|
|
||||
|
private String fromApp; |
||||
|
|
||||
|
private String fromClient; |
||||
|
|
||||
|
private Date operatingTime; |
||||
|
} |
||||
@ -0,0 +1,33 @@ |
|||||
|
package com.epmet.enums; |
||||
|
|
||||
|
import com.epmet.constant.SystemMessageType; |
||||
|
import lombok.AllArgsConstructor; |
||||
|
import lombok.Getter; |
||||
|
|
||||
|
@AllArgsConstructor |
||||
|
@Getter |
||||
|
public enum SystemMessageTypeEnum { |
||||
|
|
||||
|
POINT_RULE_CHANGED(SystemMessageType.POINT_RULE_CHANGED, "积分规则修改"); |
||||
|
|
||||
|
private String type; |
||||
|
private String typeDisplay; |
||||
|
|
||||
|
public static SystemMessageTypeEnum get(String type) { |
||||
|
for (SystemMessageTypeEnum e : SystemMessageTypeEnum.values()) { |
||||
|
if (e.type.equals(type)) { |
||||
|
return e; |
||||
|
} |
||||
|
} |
||||
|
return null; |
||||
|
} |
||||
|
|
||||
|
public static String getTypeDisplay(String type) { |
||||
|
SystemMessageTypeEnum o = get(type); |
||||
|
if (o != null) { |
||||
|
return o.getTypeDisplay(); |
||||
|
} |
||||
|
return null; |
||||
|
} |
||||
|
|
||||
|
} |
||||
Loading…
Reference in new issue