接收mqtt，处理实时数据

yangys

2024-10-08 54c90fc950538e05d328e409e504ad1a291ad180

接收mqtt，处理实时数据

 smart-man-boot/src/main/java/com/qianwen/smartman/common/config/MdcMqttConfig.java

@@ -1,6 +1,7 @@
package com.qianwen.smartman.common.config;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Random;

import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
@@ -22,10 +23,16 @@
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;


import com.alibaba.fastjson.JSONObject;
import com.qianwen.core.websocket.distribute.MessageDO;
import com.qianwen.core.websocket.distribute.RedisMessageDistributor;
import com.qianwen.smartman.common.cache.cps.WorkstationCache;
import com.qianwen.smartman.common.websocket.realtime.RealTimeDaraResponseJsonWebSocketMessage;
import com.qianwen.smartman.modules.cps.message.dto.TelemetryDataResponseDTO;

import cn.hutool.core.date.DateTime;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.json.JSONUtil;

@Configuration
public class MdcMqttConfig {
@@ -39,6 +46,9 @@
    @Value("${mqtt.password:}")
    private String mqttPassword;

    @Autowired
    private RedisMessageDistributor messageDistributor;
    
    public static final String DEFAULT_TOPIC = "mdc";
    /**
     * 反馈创建的topic（mdc中），本应用接收并处理
@@ -47,6 +57,7 @@
    
    public static final String WOCKSTATION_CREATE_TOPIC = "mdc/workstation-create";
    
    public static final String WOCKSTATION_REALTIMEDATA_TOPIC = "mdc/realtimedata";
    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
@@ -70,19 +81,23 @@
     * 
     * @return
     */
    /*
    
    @Bean
    public MessageProducer inbound() {
        java.util.Random r = new java.util.Random();
    	
        String clientId = "spring-boot-mdc-mqtt-client-inbound"+r.nextInt(1000);
    	
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter("spring-boot-mqtt-client-inbound",
                        mqttClientFactory(), COLLECT_DATA_TOPIC, FEEDBACK_TOPIC,WOCKSTATION_CREATE_TOPIC);//最后一个参数允许多个topic参数
                new MqttPahoMessageDrivenChannelAdapter(clientId,
                        mqttClientFactory(),WOCKSTATION_REALTIMEDATA_TOPIC);//最后一个参数允许多个topic参数
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }
    */
    
    
    /**
     * 入站消息处理
@@ -93,8 +108,35 @@
    public MessageHandler handler() {
        return message -> {
            String topic = String.valueOf(message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC));
            if(FEEDBACK_TOPIC.equals(topic) ) {
        	
            if(WOCKSTATION_REALTIMEDATA_TOPIC.equals(topic) ) {
                //实时数据来了，数据从collect发送来
                String payload = (String)message.getPayload();
                logger.info("接收到mqtt消息readtime,data={}",payload);//消息体如何定义，数据点名称，值，时间，用json对象传过来
                //TelemetryDataMessage result = TelemetryDataUtils.handleTelemetryDataMessage(telemetryDataMessage, workStationItem);
                JSONObject data = JSONObject.parseObject(payload);
                long workstationId = data.getLong("workstationId");
                String name = data.getString("name");//key？
                long time = data.getLong("time");
                String v = data.getString("value");
        		
                //将数据加入缓存
                TelemetryDataResponseDTO telemetryDataResponseDTO = new TelemetryDataResponseDTO(v, time);
                WorkstationCache.setWorkstationRealTime(workstationId, name, telemetryDataResponseDTO);
                
                //发送websocket消息
                RealTimeDaraResponseJsonWebSocketMessage jsonWebSocketMessage = new RealTimeDaraResponseJsonWebSocketMessage();
                jsonWebSocketMessage.setId(""+workstationId);
                jsonWebSocketMessage.setData(new HashMap<String, Object>() { // from class: org.springblade.modules.cps.message.consumer.TelemetryDataRealTimeConsumer.1
                    {
                        put(name, telemetryDataResponseDTO);
                    }
                });
                //RedisMessageDistributor messageDistributor = (RedisMessageDistributor) SpringUtil.getBean(RedisMessageDistributor.class);
                MessageDO messageDO = new MessageDO();
                messageDO.setNeedBroadcast(Boolean.FALSE);
                messageDO.setMessageText(JSONUtil.toJsonStr(jsonWebSocketMessage));
                messageDistributor.distribute(messageDO);
                
            }
        };
    }

 smart-man-boot/src/main/java/com/qianwen/smartman/common/websocket/realtime/RealTimeDataJsonMessageHandler.java

@@ -9,6 +9,9 @@
import com.qianwen.core.websocket.handler.JsonMessageHandler;
import com.qianwen.smartman.modules.coproduction.entity.OrderWorkstation;
import com.qianwen.smartman.modules.coproduction.service.IOrderWorkstationService;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.WebSocketSession;

@@ -17,6 +20,7 @@
 */
@Component
public class RealTimeDataJsonMessageHandler implements JsonMessageHandler<RealTimeDataRequestJsonWebSocketMessage> {
    private Logger log = LoggerFactory.getLogger(this.getClass());
    private final IOrderWorkstationService orderWorkstationService;

    public RealTimeDataJsonMessageHandler(final IOrderWorkstationService orderWorkstationService) {
@@ -25,13 +29,21 @@

    public void handle(WebSocketSession session, RealTimeDataRequestJsonWebSocketMessage message) {
        List<String> workstationIdList = message.getWorkstationIdList();
        List<OrderWorkstation> workstationInProcess = this.orderWorkstationService.getWorkstationInProcess(workstationIdList.stream().map(Long::valueOf).collect(Collectors.toSet()));
        Map<Long, String> orderWorkstationMap = workstationInProcess.stream().collect(Collectors.toMap(OrderWorkstation::getWorkstationId, OrderWorkstation::getOrderCode));
        log.info("收到websocket消息，message.workstationidList={}",message.getWorkstationIdList());
        //从blade_order_xxx表获取正在加工的工位，
        //List<OrderWorkstation> workstationInProcess = this.orderWorkstationService.getWorkstationInProcess(workstationIdList.stream().map(Long::valueOf).collect(Collectors.toSet()));
        
        //生成 工位id -> 工单号的map
        //Map<Long, String> orderWorkstationMap = workstationInProcess.stream().collect(Collectors.toMap(OrderWorkstation::getWorkstationId, OrderWorkstation::getOrderCode));
        
        if (Func.isNotEmpty(workstationIdList)) {
            for (String workstationId : workstationIdList) {
                RealTimeDaraResponseJsonWebSocketMessage jsonWebSocketMessage = new RealTimeDaraResponseJsonWebSocketMessage();
                Map<String, Object> map = WorkstationCache.getWorkstationRealTime(workstationId);
                map.put("orderCode", Func.isEmpty(orderWorkstationMap.get(Long.valueOf(workstationId))) ? "-" : orderWorkstationMap.get(Long.valueOf(workstationId)));
                Map<String, Object> map = WorkstationCache.getWorkstationRealTime(workstationId);//TODO：这就是返回给前端的数据，我需要在这里查询,里面缓存没有，实际上就是直接查询
                if("1656819188967653378".equals(workstationId)) {
                    log.info(workstationId+"收到:"+map);
                }
                //map.put("orderCode", Func.isEmpty(orderWorkstationMap.get(Long.valueOf(workstationId))) ? "-" : orderWorkstationMap.get(Long.valueOf(workstationId)));
                jsonWebSocketMessage.setData(map);
                jsonWebSocketMessage.setId(workstationId);
                WebSocketMessageSender.send(session, jsonWebSocketMessage);

 smart-man-boot/src/main/java/com/qianwen/smartman/modules/cps/controller/MyTestController.java

@@ -7,7 +7,11 @@
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import com.qianwen.core.boot.ctrl.BladeController;
import com.qianwen.core.datascope.props.DataScopeProperties;
import com.qianwen.core.mp.intercept.QueryInterceptor;
@@ -16,12 +20,21 @@
import com.qianwen.core.secure.annotation.PreAuth;
import com.qianwen.core.tenant.annotation.NonDS;
import com.qianwen.core.tool.api.R;
import com.qianwen.core.tool.utils.SpringUtil;
import com.qianwen.core.websocket.config.WebSocketMessageSender;
import com.qianwen.core.websocket.distribute.MessageDO;
import com.qianwen.core.websocket.distribute.RedisMessageDistributor;
import com.qianwen.smartman.common.cache.cps.WorkstationCache;
import com.qianwen.smartman.common.websocket.realtime.RealTimeDaraResponseJsonWebSocketMessage;
import com.qianwen.smartman.modules.cps.message.dto.TelemetryDataResponseDTO;
import com.qianwen.smartman.modules.cps.service.CollectDeviceTypeService;
import com.qianwen.smartman.modules.cps.service.ICommonGroupService;
import com.qianwen.smartman.modules.cps.service.WorkstationDatapointsService;
import com.qianwen.smartman.modules.cps.vo.CollectDeviceTypeVO;
import com.qianwen.smartman.modules.cps.vo.WorkstationDatapointsVO;
import com.qianwen.smartman.modules.cps.vo.WorkstationGroupVO;

import cn.hutool.json.JSONUtil;

import org.apache.commons.io.IOUtils;
import org.springframework.beans.factory.annotation.Autowired;
@@ -45,12 +58,46 @@
    DataScopeProperties dsp;
    @Autowired
    QueryInterceptor qi;
    
    @GetMapping({"/dsp"})
    @ApiOperation("dsp")
    public R<String> showdsp() {
        return R.data(dsp.toString()+qi);
        Properties props = System.getProperties();
        String appName = props.getProperty("spring.application.name");
        return R.data(appName);
    }
    
    @GetMapping({"/realtimesend"})
    @ApiOperation("realtimesend")
    public R<String> realtimesend(String wid,String k,String v) {
        RedisMessageDistributor messageDistributor = (RedisMessageDistributor) SpringUtil.getBean(RedisMessageDistributor.class);
    	
        RealTimeDaraResponseJsonWebSocketMessage jsonWebSocketMessage = new RealTimeDaraResponseJsonWebSocketMessage();
        jsonWebSocketMessage.setId(wid);
        
        Map<String, Object> map = WorkstationCache.getWorkstationRealTime(wid);
        map.keySet().forEach(key->{
            if(key.equals(k)) {
                Object d = map.get(k);
        		
                TelemetryDataResponseDTO a = (TelemetryDataResponseDTO)d;
                a.setV(v);
                System.out.println(d);
            }
        });
        jsonWebSocketMessage.setData(map);
        
        MessageDO messageDO = new MessageDO();
        messageDO.setNeedBroadcast(false);
        messageDO.setMessageText(JSONUtil.toJsonStr(jsonWebSocketMessage));
        messageDistributor.distribute(messageDO);
        
   
        return R.data("1");
    }
    
    
    
    @GetMapping({"/listDatapointsByWorkstationId"})
    @ApiOperation("根据工位id获取数据点")
    public R<String> listDatapointsByWorkstationId() {

 smart-man-boot/src/main/java/com/qianwen/smartman/modules/cps/message/consumer/TelemetryDataRealTimeConsumer.java

@@ -46,7 +46,8 @@
                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);
                    WorkstationCache.setWorkstationRealTime(workStationItem.getWorkstationId(), result.getName(), telemetryDataResponseDTO);//设置实时数据里面的缓存数据到redis
                    
                    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

 smart-man-boot/src/main/java/com/qianwen/smartman/modules/cps/service/impl/CommonGroupServiceImpl.java

@@ -720,9 +720,7 @@
        if (CollectionUtil.isEmpty(result)) {
            return new ArrayList<>();
        }
        Set<Long> groupIdSet = groupList.stream().map((v0) -> {
            return v0.getId();
        }).collect(Collectors.toSet());
        Set<Long> groupIdSet = groupList.stream().map(CommonGroup::getId).collect(Collectors.toSet());
        Iterator<MdcWorkStationGroupVO> ite = result.iterator();
        while (ite.hasNext()) {
            if (!groupIdSet.contains(ite.next().getGroupId())) {

 smart-man-boot/src/main/java/com/qianwen/smartman/modules/cps/service/impl/WorkstationServiceImpl.java

@@ -22,6 +22,7 @@
import javax.servlet.http.HttpServletResponse;

import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.jetbrains.annotations.NotNull;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
@@ -29,6 +30,8 @@
import org.springframework.web.multipart.MultipartFile;

import com.alibaba.excel.write.merge.AbstractMergeStrategy;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
@@ -96,6 +99,7 @@
import com.qianwen.smartman.modules.cps.service.IWorkstationService;
import com.qianwen.smartman.modules.cps.service.IWorkstationWcsService;
import com.qianwen.smartman.modules.cps.service.IWorkstationWorkbenchService;
import com.qianwen.smartman.modules.cps.service.WorkstationDatapointsService;
import com.qianwen.smartman.modules.cps.vo.DmpVariablesVO;
import com.qianwen.smartman.modules.cps.vo.FmsWorkstationGroupVO;
import com.qianwen.smartman.modules.cps.vo.FmsWorkstationQueryVO;
@@ -105,6 +109,7 @@
import com.qianwen.smartman.modules.cps.vo.RealTimeStatusVO;
import com.qianwen.smartman.modules.cps.vo.WorkMachineEasyVO;
import com.qianwen.smartman.modules.cps.vo.WorkstationChangeCalendarVO;
import com.qianwen.smartman.modules.cps.vo.WorkstationDatapointsVO;
import com.qianwen.smartman.modules.cps.vo.WorkstationOfWorkbenchVO;
import com.qianwen.smartman.modules.cps.vo.WorkstationPageQueryVO;
import com.qianwen.smartman.modules.cps.vo.WorkstationRealTimeStatusVO;
@@ -126,27 +131,48 @@
import com.qianwen.smartman.modules.resource.builder.oss.OssBuilder;
import com.qianwen.smartman.modules.tdengine.service.IWorkstationDynamicCollectService;

import cn.hutool.core.util.ObjectUtil;

@Service
public class WorkstationServiceImpl extends BaseServiceImpl<WorkstationMapper, Workstation> implements IWorkstationService {
    private static final Pattern CHINESE = Pattern.compile("^[A-Za-z0-9]+$");
    private final IMachineService machineService;
    private final IWorkstationOfMachineService workstationOfMachineService;
    private final IWorkstationWcsService workstationWcsService;
    private final ICommonGroupOfItemService commonGroupOfItemService;
    private final ICommonGroupService commonGroupService;
    private final ICalendarService calendarService;
    private final WorkstationMapper workstationMapper;
    private final OssBuilder ossBuilder;
    private final IDmpVariablesService dmpVariablesService;
    private final ITransferDirectoryService transferDirectoryService;
    private final ITransferDirectoryGroupService transferDirectoryGroupService;
    private final IWorkstationWorkbenchService workbenchService;
    private final IDncWsRelationTdService wsRelationTdService;
    private final IDncFixedSpaceService dncFixedSpaceService;
    @Autowired
    private IMachineService machineService;
    @Autowired
    private IWorkstationOfMachineService workstationOfMachineService;
    @Autowired
    private IWorkstationWcsService workstationWcsService;
    @Autowired
    private ICommonGroupOfItemService commonGroupOfItemService;
    @Autowired
    private ICommonGroupService commonGroupService;
    @Autowired
    private ICalendarService calendarService;
    @Autowired
    private WorkstationMapper workstationMapper;
    @Autowired
    private OssBuilder ossBuilder;
    @Autowired
    private IDmpVariablesService dmpVariablesService;
    @Autowired
    private ITransferDirectoryService transferDirectoryService;
    @Autowired
    private ITransferDirectoryGroupService transferDirectoryGroupService;
    @Autowired
    private IWorkstationWorkbenchService workbenchService;
    @Autowired
    private IDncWsRelationTdService wsRelationTdService;
    @Autowired
    private IDncFixedSpaceService dncFixedSpaceService;
    @Autowired
    private InsertWorkstationProducer workstationProducer;
    @Autowired
    private IWorkstationDynamicCollectService dynamicCollectService;
    @Autowired
    private MqttMessageSender mqttMsgSender;
    @Autowired
    private WorkstationDatapointsService wsDpService;
    
    private final IWorkstationDynamicCollectService dynamicCollectService;
    private final String NAME = "默认工作台";
    private final Integer SORT = 1;
    private final Integer DEFAULT = 1;
@@ -154,9 +180,8 @@
    private final String ALL_WORKSTATION_GROUP = CommonGroupConstant.ALL_NAME;
    private final Integer FTP_CORRELATION_DIRECTORY = 1;
    private final Integer FTP_SAME_DIRECTORY = 2;
    @Autowired
    private MqttMessageSender mqttMsgSender;

    
    /*
    public WorkstationServiceImpl(final IMachineService machineService, final IWorkstationOfMachineService workstationOfMachineService, final IWorkstationWcsService workstationWcsService, final ICommonGroupOfItemService commonGroupOfItemService, final ICommonGroupService commonGroupService, final ICalendarService calendarService, final WorkstationMapper workstationMapper, final OssBuilder ossBuilder, final IDmpVariablesService dmpVariablesService, final ITransferDirectoryService transferDirectoryService, final ITransferDirectoryGroupService transferDirectoryGroupService, final IWorkstationWorkbenchService workbenchService, final IDncWsRelationTdService wsRelationTdService, final IDncFixedSpaceService dncFixedSpaceService,final IWorkstationDynamicCollectService dynamicCollectService) {
        this.machineService = machineService;
        this.workstationOfMachineService = workstationOfMachineService;
@@ -175,6 +200,7 @@
        //this.workstationProducer = workstationProducer;
        this.dynamicCollectService = dynamicCollectService;
    }
    */

    public static Long getTimeDifference(LocalDateTime startTime, LocalDateTime endTime) {
        if (Func.isEmpty(startTime)) {
@@ -628,15 +654,19 @@

    public Map<Long, List<RealTimeStatusVO>> getRealTimeProperties(List<Long> workstationIdList) {
        Map<Long, List<RealTimeStatusVO>> properties = new HashMap<>();
        List<DmpVariablesVO> list = this.dmpVariablesService.getDmpVariablesByWorkstationIds(workstationIdList);
        if (Func.isNotEmpty(list)) {
        //在这里！！！！
        //顶顶顶顶
        //List<DmpVariablesVO> list = this.dmpVariablesService.getDmpVariablesByWorkstationIds(workstationIdList);
        //this.wsDpService.getDatapointsByWorkstationIds();
        //if (Func.isNotEmpty(list)) {
            /*
            Map<Long, List<DmpVariablesVO>> dmpVariablesMap =list.stream().collect(Collectors.groupingBy((v0) -> {
                return v0.getWorkstationId();
            }));
            dmpVariablesMap.forEach((k, v) -> {
            dmpVariablesMap.forEach((workstationId, dmpVariablesVOList) -> {
                ArrayList<RealTimeStatusVO> arrayList = new ArrayList<>();
                Map<String, Object> map = WorkstationCache.getWorkstationRealTime(String.valueOf(k));
                v.forEach(dmpVariablesVO -> {
                Map<String, Object> map = WorkstationCache.getWorkstationRealTime(String.valueOf(workstationId));//应该有数据，因为缓存没有，直接查的queryLastParameter
                dmpVariablesVOList.forEach(dmpVariablesVO -> {
                    RealTimeStatusVO vo = new RealTimeStatusVO();
                    vo.setKey(dmpVariablesVO.getName());
                    vo.setName(dmpVariablesVO.getDescription());
@@ -651,9 +681,56 @@
                    }
                    arrayList.add(vo);
                });
                properties.put(k, arrayList);
                properties.put(workstationId, arrayList);
            });
        }
            */
        	
        //}
        //yangys修改，改为数据点位的方式
        for(Long workstationId: workstationIdList) {
    		
            Map<String, Object> map = WorkstationCache.getWorkstationRealTime(String.valueOf(workstationId));//应该有数据，因为缓存没有，内部直接查的queryLastParameter
    		
            JSONObject cfgRow;
            ArrayList<RealTimeStatusVO> arrayList = new ArrayList<>();
    		
            WorkstationDatapointsVO  dpVO = wsDpService.getDatapoints(workstationId);
            if(dpVO!=null && ObjectUtil.isNotEmpty(dpVO.getDpHead())) {
    			
                String prop,label;
                JSONArray dpCfgJson = JSONArray.parseArray(dpVO.getDpConfig());
                for(int i=0;i<dpCfgJson.size();i++) {
                    cfgRow = dpCfgJson.getJSONObject(i);
	    			
                    RealTimeStatusVO vo = new RealTimeStatusVO();
                    prop = cfgRow.getString("dpName");
                    vo.setKey(prop);
	                
                    label = prop;
                    if(StringUtils.equals("DeviceStatus", prop)) {
                        label = "状态";
                    }else if(StringUtils.equals("Output", prop)) {
                        label = "产量";
                    }else if(StringUtils.equals("AlarmNo", prop)) {
                        label = "告警号";
                    }
                    vo.setName(label);
                    //vo.setUnit(headRow.getString("dpUnit"));
                    vo.setSort(i);
                    //vo.setWcsDataType(dmpVariablesVO.getWcsDataType());
                    vo.setBigScreen(false);
                    vo.setRealTimeData(true);
	                
                    if (map.containsKey(vo.getKey())) {
                        TelemetryDataResponseDTO t = (TelemetryDataResponseDTO) MapUtils.getObject(map, vo.getKey());
                        vo.setValue(t.getV());
                    }
	                
                    arrayList.add(vo);
                }
            }
            properties.put(workstationId, arrayList);
        }
        return properties;
    }

@@ -1194,9 +1271,11 @@
        if (Func.isEmpty(list)) {
            return dataList;
        }
        //TODO，格式太差
        Map<Long, List<RealTimeStatusVO>> realTimePropertiesMap = getRealTimeProperties( list.stream().map((v0) -> {
            return v0.getId();
        }).collect(Collectors.toList()));
        
        return list.stream().filter(i -> {
            boolean filter = Boolean.TRUE.booleanValue();
            if (Func.isEmpty(i.getMachineId())) {

 smart-man-boot/src/main/java/com/qianwen/smartman/modules/mdc/jobHandler/CacheNewCollectXxlJob.java

@@ -20,7 +20,8 @@
        this.collectService = collectService;
    }

    @XxlJob("allDmpCollectDataJobHandler")
    //暂时去掉，没有用
    //@XxlJob("allDmpCollectDataJobHandler")
    public ReturnT<String> allDmpCollectDataJobHandler(String param) throws Exception {
        XxlJobLogger.log("XXL-JOB, 获取DMP采集参数当前时间全量数据.....缓存构建开始", new Object[0]);
        this.collectService.cacheNowCollectData();

 smart-man-boot/src/main/java/com/qianwen/smartman/modules/mdc/mapper/SuperAggregateStateFeedbackMapper.java

@@ -15,7 +15,7 @@
@InterceptorIgnore(tenantLine = FmsConstant.AUTOMATIC)
public interface SuperAggregateStateFeedbackMapper {
    /**
     * 按生产日期获取反馈状态表的状态数据,日期参数都该为LocalDateTime
     * 按生产日期获取反馈状态表的状态数据,日期参数都该为int
     * @param workstationIds
     * @param startDate
     * @param endDate
@@ -25,11 +25,11 @@
    //List<SuperAggregateState> getStatusDataByFactoryDate(@Param("workstationIds") List<Long> workstationIds, @Param("startDate") String startDate, @Param("endDate") String endDate);

    /**
     * 获取反馈状态表的状态数据,日期参数都该为LocalDateTime
     * 获取反馈状态表的状态数据,日期参数都该为Date
     * @param workstationIds
     * @param startDate
     * @param endDate
     * @return
     */
    List<SuperAggregateState> getStatusData(@Param("workstationIds") List<Long> workstationIds, @Param("startDate") LocalDateTime startDate, @Param("endDate") LocalDateTime endDate);
    List<SuperAggregateState> getStatusData(@Param("workstationIds") List<Long> workstationIds, @Param("startDate") java.util.Date startDate, @Param("endDate") java.util.Date endDate);
}

 smart-man-boot/src/main/java/com/qianwen/smartman/modules/mdc/mapper/SuperAggregateStateMapper.java

@@ -3,8 +3,7 @@
import com.baomidou.dynamic.datasource.annotation.DS;
import com.baomidou.mybatisplus.annotation.InterceptorIgnore;

import java.time.LocalDate;
import java.time.LocalDateTime;

import java.util.Date;
import java.util.List;
import org.apache.ibatis.annotations.Param;
@@ -86,7 +85,7 @@
     * @param endDate 截至时间
     * @return 数据列表
     */
    List<SuperAggregateState> getStatusData(@Param("workstationIds") List<Long> workstationIds, @Param("startDate") LocalDateTime startDate, @Param("endDate") LocalDateTime endDate);
    List<SuperAggregateState> getStatusData(@Param("workstationIds") List<Long> workstationIds, @Param("startDate") java.util.Date startDate, @Param("endDate") java.util.Date endDate);

    /**
     * 根据工位id列表和时间节点获取状态数据yys,首页稼动率时使用的
@@ -95,7 +94,7 @@
     * @param endDate
     * @return
     */
    List<SuperAggregateState> getStatusDataByTimeSection(@Param("workstationIds") List<Long> workstationIds, @Param("startDate") LocalDateTime startDate, @Param("endDate") LocalDateTime endDate);
    List<SuperAggregateState> getStatusDataByTimeSection(@Param("workstationIds") List<Long> workstationIds, @Param("startDate") java.util.Date startDate, @Param("endDate") java.util.Date endDate);

    /**
     * 根据工厂日期获取聚合状态数据
@@ -117,7 +116,7 @@
     * @return
     */
    //List<SuperAggregateState> getEquipmentStatusDuration(@Param("workstationIds") List<Long> workstationIds, @Param("startDate") String startDate, @Param("endDate") String endDate, @Param("status") Integer status);
    List<SuperAggregateState> getEquipmentStatusDuration(@Param("workstationIds") List<Long> workstationIds, @Param("startDate") LocalDateTime startDate, @Param("endDate") LocalDateTime endDate, @Param("status") Integer status);
    List<SuperAggregateState> getEquipmentStatusDuration(@Param("workstationIds") List<Long> workstationIds, @Param("startDate") java.util.Date startDate, @Param("endDate") java.util.Date endDate, @Param("status") Integer status);

    List<SuperAggregateState> getEquipmentStatusRecord(@Param("workstationIds") List<Long> workstationIds, @Param("shiftIndex") Integer shiftIndex, @Param("factoryDate") Integer factoryDate);


 smart-man-boot/src/main/java/com/qianwen/smartman/modules/mdc/service/impl/SuperAggregateStateServiceImpl.java

@@ -20,6 +20,7 @@
import org.springframework.stereotype.Service;

import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.qianwen.core.tool.utils.DateTimeUtil;
import com.qianwen.core.tool.utils.DateUtil;
import com.qianwen.core.tool.utils.Func;
import com.qianwen.smartman.common.constant.CommonConstant;
@@ -62,7 +63,7 @@
            LocalDateTime startTime = LocalDateTime.of(startDate, LocalTime.MIN);
            LocalDateTime endTime = LocalDateTime.of(endDate.plusDays(1L), LocalTime.MIN);
            //statusDataList = this.baseMapper.getStatusData(workstationIds, startDate,endDate.plusDays(1L));
            statusDataList = this.baseMapper.getStatusData(workstationIds, startTime,endTime);
            statusDataList = this.baseMapper.getStatusData(workstationIds, DateTimeUtil.toDate(startTime),DateTimeUtil.toDate(endTime));
        }
        return buildDuration(statusDataList);
    }
@@ -70,8 +71,6 @@
    @Override
    public List<SuperAggregateState> getStatusDataWithFeedback(List<Long> workstationIds, StatisticalMethodEnum statisticalMethod, LocalDate startDate, LocalDate endDate) {
        //稼动率查询
    	
         //LocalDate startTime = LocalDateTime, LocalDate endDate
         
        List<SuperAggregateState> statusDataList;
        if (StatisticalMethodEnum.SHIFT.equals(statisticalMethod) || StatisticalMethodEnum.DAY.equals(statisticalMethod) || StatisticalMethodEnum.WEEK.equals(statisticalMethod) || StatisticalMethodEnum.MONTH.equals(statisticalMethod)) {
@@ -83,7 +82,7 @@
            //statusDataList = this.aggregateStateFeedbackMapper.getStatusData(workstationIds, LocalDateTimeUtil.format(startDate, DateConstant.PATTERN_DATE_TIME), LocalDateTimeUtil.format(endDate.plusDays(1L), DateConstant.PATTERN_DATE_TIME));
            LocalDateTime startTime = LocalDateTime.of(startDate, LocalTime.MIN);
            LocalDateTime endTime = LocalDateTime.of(endDate.plusDays(1L), LocalTime.MIN);
            statusDataList = this.aggregateStateFeedbackMapper.getStatusData(workstationIds, startTime,endTime);
            statusDataList = this.aggregateStateFeedbackMapper.getStatusData(workstationIds, DateTimeUtil.toDate(startTime),DateTimeUtil.toDate(endTime));
        }
        return buildDuration(statusDataList);
    }
@@ -91,14 +90,15 @@
    @Override
    public List<SuperAggregateState> getStatusByCondition(List<Long> workstationIds, LocalDateTime startTime, LocalDateTime endTime) {
        //List<SuperAggregateState> statusDataList = this.baseMapper.getStatusData(workstationIds, LocalDateTimeUtil.format(startTime, DateConstant.PATTERN_DATE_TIME), LocalDateTimeUtil.format(endTime, DateConstant.PATTERN_DATE_TIME));
        List<SuperAggregateState> statusDataList = this.baseMapper.getStatusData(workstationIds, startTime,endTime);
        List<SuperAggregateState> statusDataList = this.baseMapper.getStatusData(workstationIds,  DateTimeUtil.toDate(startTime), DateTimeUtil.toDate(endTime));
        return buildDuration(statusDataList);
    }

    @Override
    public List<SuperAggregateState> getStatusByTimeSection(List<Long> workstationIds, LocalDateTime startTime, LocalDateTime endTime) {
        //List<SuperAggregateState> statusDataList = this.baseMapper.getStatusDataByTimeSection(workstationIds, LocalDateTimeUtil.format(startTime, DateConstant.PATTERN_DATE_TIME), LocalDateTimeUtil.format(endTime, DateConstant.PATTERN_DATE_TIME));
        List<SuperAggregateState> statusDataList = this.baseMapper.getStatusDataByTimeSection(workstationIds, startTime, endTime);
    	
        List<SuperAggregateState> statusDataList = this.baseMapper.getStatusDataByTimeSection(workstationIds, DateTimeUtil.toDate(startTime), DateTimeUtil.toDate(endTime));
        return buildDuration(startTime, statusDataList);
    }

@@ -121,8 +121,7 @@
        //Date endTimeD = LocalDateUtil.localDateTimeToDate(endTime);
        //endTimeD = new Date(124,8,3);
        
        
        List<SuperAggregateState> equipmentStatusDuration = this.baseMapper.getEquipmentStatusDuration(workstationIds, startTime, endTime, status);
        List<SuperAggregateState> equipmentStatusDuration = this.baseMapper.getEquipmentStatusDuration(workstationIds, DateTimeUtil.toDate(startTime), DateTimeUtil.toDate(endTime), status);
        
        List<SuperAggregateState> equipmentStatusDuration2 = FilterOffUtils.filterOffDay(equipmentStatusDuration, OpenTypeEnums.TIME_USED_ANALYSIS);
        Map<Long, Long> timeMap = buildDuration(startTime, equipmentStatusDuration2).stream().collect(Collectors.groupingBy(SuperAggregate::getWorkstationId, Collectors.summingLong(SuperAggregateState::getDurationCollect))).entrySet().stream().sorted(Map.Entry.comparingByValue(Comparator.reverseOrder())).limit(top.intValue()).collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue, (oldValue, newValue) -> oldValue, java.util.LinkedHashMap::new));
@@ -154,7 +153,7 @@
        //Date startTimeD = LocalDateUtil.localDateTimeToDate(startTime);
        //Date endTimeD = LocalDateUtil.localDateTimeToDate(endTime);
        //List<SuperAggregateState> statusDataList = this.baseMapper.getEquipmentStatusDuration(workStationIdList, LocalDateTimeUtil.format(startTime, DateConstant.PATTERN_DATE_TIME), LocalDateTimeUtil.format(endTime, DateConstant.PATTERN_DATE_TIME), null);
        List<SuperAggregateState> statusDataList = this.baseMapper.getEquipmentStatusDuration(workStationIdList, startTime, endTime, null);
        List<SuperAggregateState> statusDataList = this.baseMapper.getEquipmentStatusDuration(workStationIdList, DateTimeUtil.toDate(startTime), DateTimeUtil.toDate(endTime), null);
        return buildDuration(startTime, FilterOffUtils.filterOffDay(statusDataList, OpenTypeEnums.OEE));
    }


 smart-man-boot/src/main/java/com/qianwen/smartman/modules/mdc/service/impl/WorkstationAnalysisServiceImpl.java

@@ -43,6 +43,7 @@
import com.qianwen.smartman.modules.mdc.entity.WorkstationCollectData;
import com.qianwen.smartman.modules.mdc.mapper.SuperAggregateStateMapper;
import com.qianwen.smartman.modules.mdc.mapper.SuperAlarmMapper;
import com.qianwen.smartman.modules.mdc.mapper.SuperProcessParameterMapper;
import com.qianwen.smartman.modules.mdc.service.IProcessParameterService;
import com.qianwen.smartman.modules.mdc.service.IWorkstationAnalysisService;
import com.qianwen.smartman.modules.mdc.vo.AlarmAnalysisWorkstationVO;
@@ -63,21 +64,11 @@
    private  IGlobalWcsService globalWcsService;
    @Autowired
    private  WorkstationMapper workstationMapper;
    //private final SuperProcessParameterMapper superProcessParameterMapper;
    @Autowired
    private IWorkstationService workstationService;
    @Autowired
    private IProcessParameterService processParameterService;
    /*
    public WorkstationAnalysisServiceImpl(final SuperAggregateStateMapper superAggregateStateMapper, final SuperAlarmMapper superAlarmMapper, final IGlobalWcsService globalWcsService, final WorkstationMapper workstationMapper, final SuperProcessParameterMapper superProcessParameterMapper, final IWorkstationService workstationService,IProcessParameterService aProcessParameterService) {
        this.superAggregateStateMapper = superAggregateStateMapper;
        this.superAlarmMapper = superAlarmMapper;
        this.globalWcsService = globalWcsService;
        this.workstationMapper = workstationMapper;
        this.superProcessParameterMapper = superProcessParameterMapper;
        this.workstationService = workstationService;
        this.processParameterService = aProcessParameterService;
    }*/


    @Override 
    public List<StatusAnalysisWorkstationVO> statusByWorkstation(String workstationId) {
@@ -200,7 +191,9 @@
        
        //deviceMap key为wcs的code值，value是设备数量（该状态的）
        Map<String, Long> deviceMap = wIds.stream().map(wId -> {
            Map<String, Object> realTimeData = WorkstationCache.getWorkstationAllCollect(wId);
        	
            //下面的缓存是TelemetryDataRealTimeConsumer中设置的（接收数据）。需要再collect中设置，避免格式问题，我们不需要这么干了，直接查询
            Map<String, Object> realTimeData = WorkstationCache.getWorkstationAllCollect(wId); 
            //WorkstationWcsDmpDTO dmpDTO = WorkstationCache.getDmpStatus(wId);//这里使用了blade_workstation_wcs blade_dmp_variables表，考虑去掉了，我们没有dmp
            WorkstationWcsDmpDTO dmpDTO = new WorkstationWcsDmpDTO();
            dmpDTO.setName("DeviceStatus");
@@ -210,10 +203,9 @@
                return dto.getV();
            }
            
            //如果缓存没有数据
            //this error
            //如果以上从缓存没有拿到数据(我们肯定没有缓存，不知道缓存哪儿来的),查询最新时序数据
            //List<WorkstationCollectData> workstationCollectData = this.superProcessParameterMapper.queryLastParameter(wId);//该工位最新的last(ts),last(v)
            

            List<WorkstationCollectData> workstationCollectData = processParameterService.queryLastParameter(Long.parseLong(wId));
            if (Func.isNotEmpty(workstationCollectData)) {
                //resultMapping ,key:n ，value：TelemetryDataResponseDTO{last(ts),last(v)}

 smart-man-boot/src/main/java/com/qianwen/smartman/modules/system/service/impl/DataScopeManagerServiceImpl.java

@@ -21,6 +21,7 @@
import com.qianwen.smartman.modules.cps.service.IEmployeeService;
import com.qianwen.smartman.modules.system.entity.Dept;
import com.qianwen.smartman.modules.system.entity.DeptScope;
import com.qianwen.smartman.modules.system.entity.UserDept;
import com.qianwen.smartman.modules.system.mapper.DeptMapper;
import com.qianwen.smartman.modules.system.service.IDataScopeManagerService;
import com.qianwen.smartman.modules.system.service.IDeptScopeService;
@@ -152,9 +153,7 @@
    }

    private void checkUsingDept(List<Long> deptId) {
        if (this.userDeptService.count(Lambda.in((v0) -> {
            return v0.getDeptId();
        }, deptId)) != 0) {
        if (this.userDeptService.count(Lambda.in(UserDept::getDeptId, deptId)) != 0) {
            throw new ServiceException("无法删除,账号绑定了该权限");
        }
        

 smart-man-boot/src/main/resources/application-dev.yml

@@ -8,10 +8,10 @@
# database config
datasource:
  type: mysql
  url: localhost
  url: 120.46.212.231
  port: 3306
  database: blade_boot
  username: root
  database: smart_boot
  username: qwmdc
  password: Kknd_1234
  driver-class-name: com.mysql.cj.jdbc.Driver