forked from luyan/epmet-cloud-lingshan
zxc
3 years ago
11 changed files with 290 additions and 9 deletions
@ -0,0 +1,120 @@ |
|||
package com.epmet.mq.listener.listener; |
|||
|
|||
import com.alibaba.fastjson.JSON; |
|||
import com.epmet.auth.constants.AuthOperationEnum; |
|||
import com.epmet.commons.rocketmq.constants.MQUserPropertys; |
|||
import com.epmet.commons.rocketmq.messages.CheckMQMsg; |
|||
import com.epmet.commons.rocketmq.messages.LoginMQMsg; |
|||
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.entity.LogOperationEntity; |
|||
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 zxc |
|||
* @Description 查询明文或者导出操作日志监听器 |
|||
*/ |
|||
public class CheckAndExportOperationLogListener 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) { |
|||
String tags = messageExt.getTags(); |
|||
String msg = new String(messageExt.getBody()); |
|||
String pendingMsgLabel = messageExt.getUserProperty(MQUserPropertys.BLOCKED_MSG_LABEL); |
|||
logger.info("查询明文或者导出操作日志监听器-收到消息内容:{}", msg); |
|||
CheckMQMsg msgObj = JSON.parseObject(msg, CheckMQMsg.class); |
|||
|
|||
//获取操作人信息
|
|||
OperatorInfo operatorInfo = LogOperationHelper.getInstance().getOperatorInfo(msgObj.getUserId()); |
|||
LogOperationEntity logEntity = new LogOperationEntity(); |
|||
logEntity.setCategory(messageExt.getTopic()); |
|||
logEntity.setType(tags); |
|||
logEntity.setTypeDisplay(AuthOperationEnum.getDisplay(tags)); |
|||
logEntity.setIp(msgObj.getIp()); |
|||
logEntity.setFromApp(msgObj.getFromApp()); |
|||
logEntity.setFromClient(msgObj.getFromClient()); |
|||
logEntity.setTargetId(""); |
|||
logEntity.setCustomerId(operatorInfo.getCustomerId()); |
|||
logEntity.setOperatorId(msgObj.getUserId()); |
|||
logEntity.setOperatorName(operatorInfo.getName()); |
|||
logEntity.setOperatorMobile(operatorInfo.getMobile()); |
|||
logEntity.setOperatingTime(msgObj.getOperateTime()); |
|||
logEntity.setContent(msgObj.getContent()); |
|||
|
|||
DistributedLock distributedLock = null; |
|||
RLock lock = null; |
|||
try { |
|||
distributedLock = SpringContextUtils.getBean(DistributedLock.class); |
|||
lock = distributedLock.getLock(String.format("lock:auth_operation_log:%s:%s", logEntity.getType(), logEntity.getOperatorId()), |
|||
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); |
|||
} |
|||
|
|||
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("【登录操作事件监听器】删除pendingMsgLabel成功:{}", pendingMsgLabel);
|
|||
} |
|||
} |
|||
@ -0,0 +1,28 @@ |
|||
package com.epmet.commons.rocketmq.messages; |
|||
|
|||
import lombok.Data; |
|||
|
|||
import java.util.Date; |
|||
|
|||
@Data |
|||
public class CheckMQMsg { |
|||
|
|||
/** |
|||
* 谁登录 |
|||
*/ |
|||
private String userId; |
|||
|
|||
private String content; |
|||
|
|||
/** |
|||
* 操作时间 |
|||
*/ |
|||
private Date operateTime; |
|||
|
|||
private String ip; |
|||
|
|||
private String fromApp; |
|||
|
|||
private String fromClient; |
|||
|
|||
} |
|||
@ -0,0 +1,25 @@ |
|||
package com.epmet.dto.form; |
|||
|
|||
import lombok.Data; |
|||
|
|||
import javax.validation.constraints.NotBlank; |
|||
import java.io.Serializable; |
|||
|
|||
/** |
|||
* @Author zxc |
|||
* @DateTime 2022/10/17 13:42 |
|||
* @DESC |
|||
*/ |
|||
@Data |
|||
public class DetailByTypeFormDTO implements Serializable { |
|||
|
|||
private static final long serialVersionUID = 8190270734257503603L; |
|||
|
|||
public interface DetailByTypeForm{} |
|||
|
|||
@NotBlank(message = "type不能为空",groups = DetailByTypeForm.class) |
|||
private String type; |
|||
|
|||
@NotBlank(message = "id不能为空",groups = DetailByTypeForm.class) |
|||
private String id; |
|||
} |
|||
@ -0,0 +1,20 @@ |
|||
package com.epmet.dto.result; |
|||
|
|||
import lombok.Data; |
|||
|
|||
import java.io.Serializable; |
|||
|
|||
/** |
|||
* @Author zxc |
|||
* @DateTime 2022/10/17 13:41 |
|||
* @DESC |
|||
*/ |
|||
@Data |
|||
public class DetailByTypeResultDTO implements Serializable { |
|||
|
|||
private static final long serialVersionUID = -8067849365791132347L; |
|||
|
|||
private String idCard; |
|||
|
|||
private String mobile; |
|||
} |
|||
Loading…
Reference in new issue