1、推送消息统一等20ms间隔

sec-beidou-fwd/src/main/java/com/imdroid/beidou_fwd/task/GXXfzForwarder.java

@@ -122,9 +122,9 @@ public class GXXfzForwarder extends Forwarder{
     }
 
     boolean checkResult() throws InterruptedException {
-        // 等待应答，最多等1s
+        // 等待应答，最多等0.5s
-        for(int i=0; i<10; i++){
+        for(int i=0; i<25; i++){
-            Thread.sleep(100);
+            Thread.sleep(20);
             if(listener.state == XFZTCPListener.STATE_OK) return true;
             else if(listener.state == XFZTCPListener.STATE_FAILED) return false;
         }

sec-beidou-fwd/src/main/java/com/imdroid/beidou_fwd/task/GZYMQTTForwarder.java

@@ -80,7 +80,7 @@ public class GZYMQTTForwarder extends Forwarder {
             logger.info("forward to GZY mqtt: {}",json);
             try {
                 if(!mqttClient.publish(topic, json)) break;
-                Thread.sleep(50);
+                Thread.sleep(20);
             } catch (Exception e1) {
                 e1.printStackTrace();
             }
@@ -119,7 +119,7 @@ public class GZYMQTTForwarder extends Forwarder {
             logger.info("forward to GZY mqtt angles: {}",json);
             try {
                 if(!mqttClient.publish(topic, json)) break;
-                Thread.sleep(10);
+                Thread.sleep(20);
             } catch (Exception e1) {
                 e1.printStackTrace();
             }

sec-beidou-fwd/src/main/java/com/imdroid/beidou_fwd/task/ZNYMQTTForwarder.java

@@ -0,0 +1,132 @@
+package com.imdroid.beidou_fwd.task;
+
+
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.imdroid.beidou_fwd.entity.GZYMQTTAngle;
+import com.imdroid.beidou_fwd.entity.GZYMQTTData;
+import com.imdroid.beidou_fwd.service.MQTTClient;
+import com.imdroid.common.util.GsonUtil;
+import com.imdroid.common.util.NumberUtils;
+import com.imdroid.secapi.dto.GnssCalcData;
+import com.imdroid.secapi.dto.GnssStatusMsg;
+import com.imdroid.secapi.dto.GnssStatusMsgMapper;
+import org.eclipse.paho.client.mqttv3.MqttException;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.annotation.EnableScheduling;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import java.time.LocalDateTime;
+import java.util.HashMap;
+import java.util.List;
+
+@Component
+@Configuration
+@EnableScheduling
+public class ZNYMQTTForwarder extends Forwarder {
+    static final String FORWARDER_NAME = "贵州交勘院MQTT";
+    @Value("${mqtt.server.brokerUrl}")
+    private String brokerUrl;
+    @Value("${mqtt.server.username}")
+    private String username;
+    @Value("${mqtt.server.password}")
+    private String password;
+    @Value("${mqtt.server.clientid}")
+    private String clientid;
+    @Value("${mqtt.server.topic}")
+    private String topic;
+
+    @Autowired
+    GnssStatusMsgMapper statusMsgMapper;
+    MQTTClient mqttClient;
+
+    @PostConstruct
+    void registerMe() throws MqttException {
+        init(FORWARDER_NAME, "MQTT "+brokerUrl,2,FWD_DEVICE_ALIAS_NAME,30);
+        mqttClient = new MQTTClient(brokerUrl, username, password,clientid);
+        mqttClient.connect();
+    }
+
+    /**
+     * 每半小时转发GNSS解算结果
+     */
+    @Scheduled(cron = "0 0 0/1 * * ?") // 每小时执行一次
+    //@Scheduled(cron = "0 0/5 * * * ?") // 每30分钟执行一次
+    private void forwardGnss() {
+        logger.info("gzy mqtt forwardGnss");
+        forwardCurrentGnss();
+        //forwardAngleData();
+    }
+
+    @Override
+    int send(String projectId, List<GnssCalcData> records, LocalDateTime sentTime) {
+        int sendNum = 0;
+
+        for (GnssCalcData locationRecord : records) {
+            GZYMQTTData tranData = new GZYMQTTData();
+            tranData.setCollectTime(locationRecord.getCreatetime().format(formatter));
+            double n = NumberUtils.scale(locationRecord.getRposn(), 2);
+            double e = NumberUtils.scale(locationRecord.getRpose(), 2);
+            double d = NumberUtils.scale(locationRecord.getRposd(), 2);
+            tranData.setX(n);
+            tranData.setY(e);
+            tranData.setZ(d);
+            tranData.setDeviceType(2);
+            tranData.setDeviceSn(locationRecord.getDeviceid());
+            String json = GsonUtil.toJson(tranData);
+            logger.info("forward to GZY mqtt: {}",json);
+            try {
+                if(!mqttClient.publish(topic, json)) break;
+                Thread.sleep(20);
+            } catch (Exception e1) {
+                e1.printStackTrace();
+            }
+            sendNum++;
+        }
+
+        return sendNum;
+    }
+
+    void forwardAngleData() {
+        // 获取最近半小时的状态消息
+        LocalDateTime sentTime = LocalDateTime.now();
+        QueryWrapper<GnssStatusMsg> queryWrapper = new QueryWrapper<>();
+        queryWrapper.eq("tenantid",tenantId);
+        queryWrapper.le("createtime",sentTime);
+        queryWrapper.ge("createtime",sentTime.minusMinutes(fwdCycleMinutes));
+        queryWrapper.orderByDesc("createtime");
+        HashMap<String,String> sendDeviceMap = new HashMap<>();
+        List<GnssStatusMsg> statusMsgs = statusMsgMapper.selectList(queryWrapper);
+        int sendNum = 0;
+        for(GnssStatusMsg msg:statusMsgs){
+            if(sendDeviceMap.get(msg.getDeviceid())!=null) continue;
+            sendDeviceMap.put(msg.getDeviceid(),msg.getDeviceid());
+
+            GZYMQTTAngle tranData = new GZYMQTTAngle();
+            tranData.setDeviceSn(msg.getDeviceid());
+            tranData.setCollectTime(msg.getCreatetime().format(formatter));
+            float x = msg.getRoll()==null?0:msg.getRoll();
+            float y = msg.getPitch()==null?0:msg.getPitch();
+            float az = msg.getYaw()==null?0:msg.getYaw();
+            tranData.setAngleX(NumberUtils.scale((double) x,2));
+            tranData.setAngleY(NumberUtils.scale((double) y,2));
+            tranData.setAngleAz(NumberUtils.scale((double) az,2));
+
+            String json = GsonUtil.toJson(tranData);
+            logger.info("forward to GZY mqtt angles: {}",json);
+            try {
+                if(!mqttClient.publish(topic, json)) break;
+                Thread.sleep(20);
+            } catch (Exception e1) {
+                e1.printStackTrace();
+            }
+            sendNum++;
+        }
+        logger.info("total number of angles sent to GZY: {}",sendNum);
+
+    }
+
+}