package com.qianwen.smartman.modules.cps.message.consumer;
|
|
import cn.hutool.json.JSONUtil;
|
import java.util.HashMap;
|
import java.util.List;
|
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
|
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
|
import org.apache.rocketmq.spring.annotation.ConsumeMode;
|
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
|
import org.apache.rocketmq.spring.core.RocketMQListener;
|
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
|
import org.slf4j.Logger;
|
import org.slf4j.LoggerFactory;
|
import com.qianwen.smartman.common.cache.cps.WorkstationCache;
|
import com.qianwen.smartman.common.cache.cps.WorkstationOfMachineCache;
|
import com.qianwen.smartman.common.utils.CommonUtil;
|
import com.qianwen.smartman.common.websocket.realtime.RealTimeDaraResponseJsonWebSocketMessage;
|
import com.qianwen.core.tool.utils.Func;
|
import com.qianwen.core.tool.utils.SpringUtil;
|
import com.qianwen.core.websocket.distribute.MessageDO;
|
import com.qianwen.core.websocket.distribute.RedisMessageDistributor;
|
import com.qianwen.smartman.modules.cps.dto.WorkstationParamTypeDTO;
|
import com.qianwen.smartman.modules.cps.entity.WorkstationOfMachine;
|
import com.qianwen.smartman.modules.cps.message.dto.TelemetryDataResponseDTO;
|
import com.qianwen.smartman.modules.cps.message.event.TelemetryDataMessage;
|
import com.qianwen.smartman.modules.cps.utils.TelemetryDataUtils;
|
import org.springframework.stereotype.Component;
|
|
/**
|
* rockmq消息消费者 telemetry(遥测)
|
*/
|
@RocketMQMessageListener(topic = "telemetry-data", consumerGroup = "telemetry-data-real-time-consumer", consumeMode = ConsumeMode.ORDERLY)
|
@Component
|
public class TelemetryDataRealTimeConsumer implements RocketMQListener<TelemetryDataMessage>, RocketMQPushConsumerLifecycleListener {
|
private static final Logger log = LoggerFactory.getLogger(TelemetryDataRealTimeConsumer.class);
|
|
public void onMessage(TelemetryDataMessage telemetryDataMessage) {
|
List<WorkstationOfMachine> workStationOfMachineList = WorkstationOfMachineCache.getList(telemetryDataMessage.getMachineId());
|
if (Func.isEmpty(workStationOfMachineList)) {
|
log.info("[采集数据] 对应得采集设备{}没有绑定任何工位，直接丢弃", telemetryDataMessage);
|
return;
|
}
|
List<WorkstationParamTypeDTO> workstationParamTypeDTOList = WorkstationCache.getWorkstationParamTypeCache(telemetryDataMessage.getMachineId(), telemetryDataMessage.getName());
|
if (Func.isNotEmpty(workstationParamTypeDTOList)) {
|
workstationParamTypeDTOList.forEach(workStationItem -> {
|
final TelemetryDataMessage result = TelemetryDataUtils.handleTelemetryDataMessage(telemetryDataMessage, workStationItem);
|
if (Func.isNotEmpty(result)) {
|
final TelemetryDataResponseDTO telemetryDataResponseDTO = new TelemetryDataResponseDTO(result.getValue(), result.getTimestamp());
|
WorkstationCache.setWorkstationRealTime(workStationItem.getWorkstationId(), result.getName(), telemetryDataResponseDTO);
|
RealTimeDaraResponseJsonWebSocketMessage jsonWebSocketMessage = new RealTimeDaraResponseJsonWebSocketMessage();
|
jsonWebSocketMessage.setId(workStationItem.getWorkstationId().toString());
|
jsonWebSocketMessage.setData(new HashMap<String, Object>() { // from class: org.springblade.modules.cps.message.consumer.TelemetryDataRealTimeConsumer.1
|
{
|
put(result.getName(), telemetryDataResponseDTO);
|
}
|
});
|
RedisMessageDistributor messageDistributor = (RedisMessageDistributor) SpringUtil.getBean(RedisMessageDistributor.class);
|
MessageDO messageDO = new MessageDO();
|
messageDO.setNeedBroadcast(Boolean.FALSE);
|
messageDO.setMessageText(JSONUtil.toJsonStr(jsonWebSocketMessage));
|
messageDistributor.distribute(messageDO);
|
}
|
});
|
}
|
}
|
@Override
|
public void prepareStart(DefaultMQPushConsumer consumer) {
|
consumer.setMaxReconsumeTimes(1);
|
consumer.setInstanceName(CommonUtil.changeInstanceNameToPID(consumer));
|
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
|
}
|
|
|
}
|
