package com.qianwen.smartman.modules.dmpLog.message.consumer;
|
|
import cn.hutool.log.StaticLog;
|
import com.alibaba.fastjson.JSONObject;
|
import java.time.Instant;
|
import java.time.LocalDateTime;
|
import java.time.ZoneId;
|
import java.time.format.DateTimeFormatter;
|
import java.util.Objects;
|
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
|
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
|
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
|
import org.apache.rocketmq.spring.core.RocketMQListener;
|
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
|
import com.qianwen.core.tool.utils.Func;
|
import com.qianwen.smartman.modules.dmpLog.constant.DmpLogSyncConstant;
|
import com.qianwen.smartman.modules.dmpLog.dto.DmpLogMessage;
|
import com.qianwen.smartman.modules.dmpLog.entity.SignalFlowLog;
|
import com.qianwen.smartman.modules.dmpLog.entity.SignalLog;
|
import com.qianwen.smartman.modules.dmpLog.entity.SignalMethodTimeLog;
|
import com.qianwen.smartman.modules.dmpLog.enums.LogTypeEnum;
|
import com.qianwen.smartman.modules.dmpLog.mapper.DmpLogSignalFlowMapper;
|
import com.qianwen.smartman.modules.dmpLog.mapper.DmpLogSignalMapper;
|
import com.qianwen.smartman.modules.dmpLog.mapper.DmpLogSignalMethodTimeMapper;
|
import org.springframework.beans.factory.annotation.Autowired;
|
import org.springframework.stereotype.Component;
|
|
/**
|
* 利用rockmq，从日志中心接受日志信息存入本地库tdengine中
|
*/
|
@RocketMQMessageListener(consumerGroup = DmpLogSyncConstant.CONSUMER_GROUP, topic = DmpLogSyncConstant.DMP_LOG_TOPIC)
|
@Component
|
public class DmpLogSyncConsumer implements RocketMQListener<DmpLogMessage>, RocketMQPushConsumerLifecycleListener {
|
private DateTimeFormatter fmt = DateTimeFormatter.ofPattern("yyyyMM");
|
@Autowired
|
private DmpLogSignalMapper dmpLogSignalMapper;
|
@Autowired
|
private DmpLogSignalFlowMapper dmpLogSignalFlowMapper;
|
@Autowired
|
private DmpLogSignalMethodTimeMapper dmpLogSignalMethodTimeMapper;
|
|
public void onMessage(DmpLogMessage dmpLogMessage) {
|
JSONObject jsonObject = dmpLogMessage.getMessage();
|
if (Func.isNotEmpty(jsonObject)) {
|
Integer msgType = jsonObject.getInteger("Type");
|
try {
|
switch ((LogTypeEnum) Objects.requireNonNull(LogTypeEnum.getValue(msgType))) {
|
case Signal:
|
SignalLog signalLog = (SignalLog) jsonObject.toJavaObject(SignalLog.class);
|
signalLog.setCollectMonth(Integer.valueOf(getYearMonth(signalLog.getTimeStamp())));
|
this.dmpLogSignalMapper.save(signalLog);
|
break;
|
case Signal_Flow:
|
SignalFlowLog signalFlowLog = (SignalFlowLog) jsonObject.toJavaObject(SignalFlowLog.class);
|
signalFlowLog.setCollectMonth(Integer.valueOf(getYearMonth(signalFlowLog.getTimeStamp())));
|
this.dmpLogSignalFlowMapper.save(signalFlowLog);
|
break;
|
case HanlderInnerMethod:
|
SignalMethodTimeLog signalMethodTimeLog = (SignalMethodTimeLog) jsonObject.toJavaObject(SignalMethodTimeLog.class);
|
signalMethodTimeLog.setCollectMonth(Integer.valueOf(getYearMonth(signalMethodTimeLog.getTimeStamp())));
|
this.dmpLogSignalMethodTimeMapper.save(signalMethodTimeLog);
|
break;
|
}
|
} catch (Exception e) {
|
StaticLog.error("存储dmplog日志异常，信息为：{}", new Object[]{e});
|
}
|
}
|
}
|
|
private int getYearMonth(Long timestamp) {
|
ZoneId zone = ZoneId.systemDefault();
|
LocalDateTime localDateTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(timestamp.longValue()), zone);
|
Integer rangeParam = Integer.valueOf(localDateTime.format(this.fmt));
|
return rangeParam.intValue();
|
}
|
|
public void prepareStart(DefaultMQPushConsumer consumer) {
|
consumer.setMaxReconsumeTimes(3);
|
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
|
}
|
}
|
