上线提交新代码，增加视频拉流，增加设备

src/main/java/com/template/common/mqtt/MqttPushClient.java

@@ -41,6 +41,8 @@ public class MqttPushClient {
         //设置会话心跳时间 单位为秒 服务器会每隔(1.5*keepTime)秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
         option.setKeepAliveInterval(KeepAlive);
         //setWill方法，如果项目中需要知道客户端是否掉线可以调用该方法。设置最终端口的通知消息
+        //设置最大发布数量 默认为10
+        option.setMaxInflight(200);
         //option.setWill(topic, "close".getBytes(), 2, true);
         return option;
     }

src/main/java/com/template/common/mqtt/PushCallback.java

@@ -53,6 +53,7 @@ public class PushCallback implements MqttCallback {
         // 当连接丢失时，这个方法会被调用
         // 连接丢失后，一般在这里面进行重连
         System.out.println(TimeExchange.getTime()+"[MQTT] 连接断开，30S之后尝试重连..."+cause);
+        cause.printStackTrace();
         while (true) {
             try {
                 Thread.sleep(30000);
@@ -79,23 +80,17 @@ public class PushCallback implements MqttCallback {
     public void messageArrived(String topic, MqttMessage message) {
         // subscribe后得到的消息会执行到这里面
 
-        //说明：发布一次会返回一个设备数据，所以目前循环设备拿数据的时候得sleep十秒
-        //压力床 罗会把四个设备的数据合成一个传过来
             System.out.println(2222222);
 
         log.info("接收消息主题 : " + topic);
         log.info("接收消息Qos : " + message.getQos());
         String content = byte2HexStr(message.getPayload());
         log.info("接收消息内容 : " + content);
-        if(content.length()<74){
+        if(content.length()<66){
             return;
         }
-        Devices devices=devicesService.getOne(new QueryWrapper<Devices>().eq("name",topic.substring(0,4)));
-                //设备返回回来的数据格式规则：
-                //前4位固定 5-6代表数据长度 发布时得两倍
-                //从第7位开始 每4个字符代表一个数据
-                //最后四位是CRC用去校验，没要求可不校验
-                //回调把数据存入数据库
+        Devices devices=devicesService.getOne(new QueryWrapper<Devices>().eq("device_no",content.substring(0,2)));
+
                 //01032004b600000121018e000000000000000000000030000000000001000000000000a13f
         List<Records> recordsList=new ArrayList<>();
         String[] arr= {"Temperature","Humidity","Light","CO2","Rotate","Air"};
@@ -105,25 +100,25 @@ public class PushCallback implements MqttCallback {
             switch (arr[i]){
                 case "Temperature":
                     records.setType("Temperature");
-                    records.setValue((double) Integer.parseInt(content.substring(14,18), 16)/10);
+                    records.setValue((double) Integer.parseInt(content.substring(10,14), 16)/10);
                     records.setCreateAt(new Date());
                     records.setUpdateAt(new Date());
                     break;
                 case "Humidity":
                     records.setType("Humidity");
-                    records.setValue((double) Integer.parseInt(content.substring(18,22), 16)/10);
+                    records.setValue((double) Integer.parseInt(content.substring(6,10), 16)/10);
                     records.setCreateAt(new Date());
                     records.setUpdateAt(new Date());
                     break;
                 case "CO2":
                     records.setType("CO2");
-                    records.setValue((double) Integer.parseInt(content.substring(6,10), 16));
+                    records.setValue((double) Integer.parseInt(content.substring(14,18), 16));
                     records.setCreateAt(new Date());
                     records.setUpdateAt(new Date());
                     break;
                 case "Light":
                     records.setType("Light");
-                    records.setValue((double) Integer.parseInt(content.substring(54,58), 16));
+                    records.setValue((double) Integer.parseInt(content.substring(18,22), 16));
                     records.setCreateAt(new Date());
                     records.setUpdateAt(new Date());
                     break;

src/main/java/com/template/controller/DevicePublicController.java

@@ -13,9 +13,11 @@ import com.template.model.Devices;
 import com.template.common.result.CommonResult;
 import com.template.services.DevicesService;
 import io.swagger.annotations.ApiOperation;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.util.StringUtils;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
@@ -34,6 +36,7 @@ import java.util.TimerTask;
  * @since 2024-04-25
  */
 @RestController
+@Slf4j
 @RequestMapping("/api/devicePublic")
 public class DevicePublicController  {
 
@@ -54,21 +57,37 @@ public class DevicePublicController  {
     public CommonResult publicDeviceMethod() throws UnsupportedEncodingException {
         System.out.println("设备开始发布");
         List<Devices> devicesList =devicesService.list(new QueryWrapper<>());
+        int count =devicesList.size();
         if(devicesList != null && devicesList.size() > 0){
             //修改定时任务创建方式
             Timer[] timers=new Timer[devicesList.size()];
+//            任务计数
             int i=0;
+            //重新发布前先清空timer任务
+            if (TimerMap.timerList.size()>0){
+                for(Timer timer:TimerMap.timerList){
+                    timer.cancel();
+                }
+                TimerMap.timerList.clear();
+            }
+            long delay=1000;
+            String topicOld="";
+            String topicNew="";
             for (Devices devices:devicesList){
-                String command = devices.getDeviceSerial();
-                String topic = devices.getName()+"_GET";
-                String outTopic = devices.getName()+"_PUB";
+                String command = devices.getDeviceNo()+"030AF1000E";
+                topicOld=topicNew;
+                String topic = devices.getTopic()+"_GET";
+                topicNew=topic;
+                String outTopic = devices.getTopic()+"_PUB";
                 JSONObject jsonObject = new JSONObject();
-                if (topic==null || topic.trim().length()==0){
-                    System.out.println("topic不能为空，command："+command+"；topic："+topic);
+                if (!StringUtils.hasText(devices.getTopic())){
+                    log.info("topic不能为空，command："+devices.getDeviceNo()+"；topic："+devices.getTopic());
+                    count--;
                     continue;
                 }
-                if (command==null || command.trim().length()==0){
-                    System.out.println("command不能为空，command："+command+"；topic："+topic);
+                if (!StringUtils.hasText(devices.getDeviceNo())){
+                    System.out.println("command不能为空，command："+devices.getDeviceNo()+"；topic："+topic);
+                    count--;
                     continue;
                 }
 //                if (command.length()!=12){
@@ -79,34 +98,32 @@ public class DevicePublicController  {
                 //可能需要在这里加一次订阅
                 //mqttPushClient.subscribe("outTopic",1);//只需要订阅一次 订阅多次会返回多条数据
                 int[] Qos={1};
-                command="010310000010";
                 String CRC = CRCGenerator.getCRC16_Modbus_Str(command);
-                command = command+" "+CRC;
+                command = command+CRC;
                 //command = HexStringUtils.toHex(command.replaceAll(" ","")).replaceAll(" ", "");
                 byte[] command2 = Hex.hexStr2Str(command.replaceAll(" ",""));
-                //任务数大于或等于设备数量时清空timer任务
-                if (TimerMap.timerList.size()>=devicesList.size()){
-                    for(Timer timer:TimerMap.timerList){
-                        timer.cancel();
-                    }
-                    TimerMap.timerList.clear();
-                }
                 TimerMap.timerList.add(new Timer());
                 timers[i] = TimerMap.timerList.get(i);
                 //消息中心软件默认六分钟 所以不满足现有需求 得做定时：每五秒执行一次任务
                 TimerTask task = new TimerTask() {
                     @Override
                     public void run() {
-                        System.out.println(topic+"====="+command2+ TimeExchange.getTime());
+                        log.info(topic+"====="+command2+ TimeExchange.getTime());
                         mqttPushClient.publish2(1,true, topic,command2);
                         mqttPushClient.subscribe(outTopic,1);//只需要订阅一次 订阅多次会返回多条数据
                     }
                 };
+
+                //主题相同则设置为延时1s执行
+                if(!topicOld.equals(topicNew)){
+                    delay=1000;
+                }
                 // 每隔5秒执行一次任务
-                timers[i].schedule(task, 1000, Long.parseLong(
+                timers[i].schedule(task, delay, Long.parseLong(
                         pollTime));
-                System.out.println("发布成功，command："+command+"；topic："+topic);
-                System.out.println("非静态方法被调用");
+                //每个任务延时增加2s 执行间隔设置为30s防止并发网关获取部到数据 12*2=24
+                delay =delay+2000;
+                log.info("发布成功，command："+command+"；topic："+topic);
                 i++;
                 continue;
             }

src/main/java/com/template/controller/DevicesController.java

@@ -63,21 +63,21 @@ public class DevicesController {
             queryWrapper.eq("company_id",companyId);
         }
         Map<String,Object> map=new HashMap<>();
-
-        Configs configs=configsService.getOne(new QueryWrapper<Configs>().eq("name","ys_access_token"));
-        JSONObject jsonObject = JSONObject.parseObject(configs.getValue());
-        if(jsonObject!=null) {
-            Long time =jsonObject.getLong("expireTime");
-            long l = System.currentTimeMillis();
-            if(time<l){
-                scheduledService.updateAccessToken();
-                configs=configsService.getOne(new QueryWrapper<Configs>().eq("name","ys_access_token"));
-                jsonObject = JSONObject.parseObject(configs.getValue());
-            }
-            map.put("accessToken", jsonObject.getString("accessToken"));
-        }else {
-            map.put("accessToken", "");
-        }
+//
+//        Configs configs=configsService.getOne(new QueryWrapper<Configs>().eq("name","ys_access_token"));
+//        JSONObject jsonObject = JSONObject.parseObject(configs.getValue());
+//        if(jsonObject!=null) {
+//            Long time =jsonObject.getLong("expireTime");
+//            long l = System.currentTimeMillis();
+//            if(time<l){
+//                scheduledService.updateAccessToken();
+//                configs=configsService.getOne(new QueryWrapper<Configs>().eq("name","ys_access_token"));
+//                jsonObject = JSONObject.parseObject(configs.getValue());
+//            }
+//            map.put("accessToken", jsonObject.getString("accessToken"));
+//        }else {
+//            map.put("accessToken", "");
+//        }
         List<Devices> list = devicesService.list(queryWrapper);
 
         map.put("list",list);

src/main/java/com/template/model/Devices.java

@@ -70,5 +70,7 @@ public class Devices implements Serializable {
     @TableField("deviceSerial")
     private String deviceSerial;
 
+    private String topic;
 
+    private String deviceNo;
 }

src/main/java/com/template/scheduled/ScheduledService.java

@@ -127,8 +127,8 @@ public class ScheduledService {
     }
 
     //定时更新token
-    @Scheduled(cron="0 30 0 * * ?")
-    @RequestMapping("/updateAccessToken")
+//    @Scheduled(cron="0 30 0 * * ?")
+//    @RequestMapping("/updateAccessToken")
     public void updateAccessToken() {
         Configs configs=configsService.getOne(new QueryWrapper<Configs>().eq("name","ys_access_token"));
         RestTemplate client = new RestTemplate();
@@ -156,8 +156,8 @@ public class ScheduledService {
     }
 
     //定时更新视频地址
-    @Scheduled(cron="0 0 1 * * ?")
-    @RequestMapping("/updatePicture")
+//    @Scheduled(cron="0 0 1 * * ?")
+//    @RequestMapping("/updatePicture")
     public void updatePicture() {
         RestTemplate client = new RestTemplate();
         Configs configs=configsService.getOne(new QueryWrapper<Configs>().eq("name","ys_access_token"));
@@ -202,4 +202,132 @@ public class ScheduledService {
         }
         devicesService.saveOrUpdateBatch(devicesList);
     }
+
+    //定时更新拉流地址
+    @Scheduled(cron="0 0 1 * * ?")
+    @RequestMapping("/updatePicture")
+    public void updatePicture1() {
+        RestTemplate client = new RestTemplate();
+        Configs configs=configsService.getOne(new QueryWrapper<Configs>().eq("name","ZLMediaKit_secret"));
+        String accessToken =configs.getValue();
+        logger.info("查询accesstoken返回结果:"+accessToken);
+        List<Devices> devicesList = devicesService.list(new QueryWrapper<>());
+        for (Devices devices:devicesList){
+            String[] arr=devices.getDeviceSerial().split(",");
+            if (arr==null||arr.length<1){
+                continue;
+            }
+            StringBuffer picbuffer=new StringBuffer();
+            for (int i = 0; i <arr.length ; i++) {
+                String requestUrl=url+"/index/api/addStreamProxy?secret="+accessToken+
+                        "&vhost="+"http://nyfc.chuanghai-tech.com/liveStream"+"&app=live" +
+                        "&stream="+devices.getId()+i+
+                        "&timeout_sec=10&enable_hls=1&enable_hls_fmp4=1&enable_rtsp=1&enable_rtmp=1" +
+                        "&enable_ts=1&enable_fmp4=1&enable_audio=1&add_mute_audio=1&mp4_max_second=1800"+
+                        "&url="+arr[i];
+                String result = client.getForObject(requestUrl, String.class);
+                logger.info("获取视频链接返回结果"+result);
+                JSONObject jsonObject = JSONObject.parseObject(result);
+                String status = String.valueOf(jsonObject.get("code"));
+                if("0".equals(status)){
+                    String picUrl =jsonObject.getJSONObject("data").getString("key")+".live.mp4";
+                    if (i!=0){
+                        picbuffer.append(",");
+                    }
+                    picbuffer.append(picUrl);
+                }
+            }
+            devices.setPicture(picbuffer.toString());
+        }
+        devicesService.saveOrUpdateBatch(devicesList);
+    }
+//    拉流参数注解
+    /*
+    secret
+string
+描述：api操作密钥(配置文件配置)
+
+vhost
+string
+描述：添加的流的虚拟主机，例如__defaultVhost__
+
+app
+string
+描述：添加的流的应用名，例如live
+
+stream
+string
+描述：添加的流的id名，例如test
+
+url
+string
+描述：拉流地址，例如rtmp://live.hkstv.hk.lxdns.com/live/hks2
+
+rtp_type
+string
+描述：rtsp拉流时，拉流方式，0：tcp，1：udp，2：组播
+
+timeout_sec
+string
+描述：拉流超时时间，单位秒，float类型
+
+retry_count
+string
+描述：拉流重试次数,不传此参数或传值<=0时，则无限重试
+
+enable_hls
+string
+描述：是否转hls-ts
+
+enable_hls_fmp4
+string
+描述：是否转hls-fmp4
+
+enable_mp4
+string
+描述：是否mp4录制
+
+enable_rtsp
+string
+描述：是否转协议为rtsp/webrtc
+
+enable_rtmp
+string
+描述：是否转协议为rtmp/flv
+
+enable_ts
+string
+描述：是否转协议为http-ts/ws-ts
+
+enable_fmp4
+string
+描述：是否转协议为http-fmp4/ws-fmp4
+
+enable_audio
+string
+描述：转协议是否开启音频
+
+add_mute_audio
+string
+描述：转协议无音频时，是否添加静音aac音频
+
+mp4_save_path
+string
+描述：mp4录制保存根目录，置空使用默认目录
+
+mp4_max_second
+string
+描述：mp4录制切片大小，单位秒
+
+hls_save_path
+string
+描述：hls保存根目录，置空使用默认目录
+
+modify_stamp
+string
+描述：是否修改原始时间戳，默认值2；取值范围：0.采用源视频流绝对时间戳，不做任何改变;1.采用zlmediakit接收数据时的系统时间戳(有平滑处理);2.采用源视频流时间戳相对时间戳(增长量)，有做时间戳跳跃和回退矫正
+
+auto_close
+string
+描述：无人观看时，是否直接关闭(而不是通过on_none_reader hook返回close)*/
 }

src/main/resources/application-dev.yml

@@ -41,6 +41,11 @@ cluster:
 # mybatis-plus不属于spring
 mybatis-plus:
   mapper-locations: classpath:/mapper/template/*.xml
+  configuration:
+  #    #开启sql打印
+  #    log-impl: org.apache.ibatis.logging.stdout.StdOutImpl
+  #    #关闭sql打印
+    log-impl: org.apache.ibatis.logging.nologging.NoLoggingImpl
   global-config:
     db-config:
       logic-delete-value: 1
@@ -65,7 +70,7 @@ mybatis-plus:
 mqtt:
   port: 81
   host: tcp://101.33.214.65:1883
-  clientId: ny_mqttPro               #MQTT-连接服务器默认客户端ID
+  clientId: ny_mqttProTest               #MQTT-连接服务器默认客户端ID
   userName: admin                    #MQTT-用户名
   password: iovadmin                 #MQTT-密码
   timeout: 2                         #MQTT-超时 单位秒
@@ -75,7 +80,7 @@ mqtt:
 
 #获取监控地址
 openys7:
-  url: https://open.ys7.com/api/lapp
+  url: http://192.168.161.225:8080
   appKey: dc94b29250f6408a8bc98a84b7d407ac
   secret: 77139b53f930a5891ce38327aed7ea02
 

src/main/resources/application.yml

@@ -38,6 +38,11 @@ spring:
         max-wait: 60000s
 # mybatis-plus不属于spring
 mybatis-plus:
+  configuration:
+    #    #开启sql打印
+    #    log-impl: org.apache.ibatis.logging.stdout.StdOutImpl
+    #    #关闭sql打印
+    log-impl: org.apache.ibatis.logging.nologging.NoLoggingImpl
   mapper-locations: classpath:/mapper/template/*.xml
   global-config:
     db-config:
@@ -73,6 +78,6 @@ mqtt:
 
 #获取监控地址
 openys7:
-  url: https://open.ys7.com/api/lapp
+  url: http://100.126.51.1:8080
   appKey: dc94b29250f6408a8bc98a84b7d407ac
   secret: 77139b53f930a5891ce38327aed7ea02

src/main/resources/logback-spring.xml

@@ -31,7 +31,7 @@
         <!--<appender-ref ref="CONSOLE"/>-->
     <!--</logger>-->
 
-    <root level="info">
+    <root level="INFO">
         <appender-ref ref="FILE"/>
         <appender-ref ref="CONSOLE"/>
     </root>