ffmpeg工具修改

loadCmd.properties

@@ -1,13 +1,18 @@
 #推流地址（eg：阿里云地址）
-pushUrl=rtmp://192.168.10.101:19350/rlive/stream_12?sign=LeL5Wchx
+pushUrl=rtmp://192.168.10.101:19350/rlive/stream_11?sign=rHtBg3sz
-#¶à·ÍÆÁ÷
-pushUrlMap=tee "[f=flv]rtmp://192.168.10.101:19350/rlive/stream_12?sign=LeL5Wchx | [f=flv]rtmp://192.168.10.101:19350/rlive/stream_17?sign=64FIaU8X"
-pushUrl2=rtmp://192.168.10.101:19350/rlive/stream_17?sign=64FIaU8X
 #拉流地址（eg：吊舱拉流的地址）
-playUrl=http://192.168.10.101:18000/flv/live/34020000001110000001_34020000001320000099_0200000099.flv
+#playUrl=http://192.168.10.101:18000/flv/live/34020000001110000001_34020000001320000099_0200000099.flv
+playUrl=http://192.168.10.101:18000/flv/live/stream_176.flv
 #固定通道 1 获取通道 2   多线程双路 3  单线程多路 4
 type=4
 
 mqttUrl=tcp://106.15.120.154
-mqttTopic=/v1/123987/rtmp/live
+#主题 12345替换对应机场编号  004替换控制板id
-time=60000
+mqttTopic=/v1/12345/rtmp/live,v1/004/confirm/DroneBMS
+
+#推流时长（毫秒）1000表示1秒
+time=600000
+
+#多路推流(选配)
+pushUrlMap=tee "[f=flv]rtmp://221.226.114.142:19350/rlive/stream_9?sign=f8a15b6n | [f=flv]rtmp://221.226.114.142:19350/rlive/stream_11?sign=rHtBg3sz"
+pushUrl2=rtmp://192.168.10.101:19350/rlive/stream_17?sign=64FIaU8X

loadFFmpeg.properties

@@ -8,6 +8,6 @@ callback=http://127.0.0.1/callback
 timeout=300
 
 #开启保活线程
-keepalive=false
+keepalive=true
 #是否输出debug消息
 debug=true

src/main/java/com/github/bluesbruce/ffch/handler/DefaultOutHandlerMethod.java

@@ -18,6 +18,7 @@ public class DefaultOutHandlerMethod implements OutHandlerMethod{
 	
 	@Override
 	public void parse(String id,String msg) {
+		long begin = System.currentTimeMillis();
 		//过滤消息
 		if (msg.indexOf("fail") != -1) {
 			log.info(id + "任务可能发生故障：" + msg);
@@ -31,6 +32,10 @@ public class DefaultOutHandlerMethod implements OutHandlerMethod{
 			log.info(id + "任务可能发生丢包10054：" + msg);
 			log.info("失败，设置中断状态");
 			isBroken=false;
+		}else if(msg.indexOf("Invalid")!= -1) {
+			log.info(id + "任务可能发生非法响应：" + msg);
+			log.info("失败，设置中断状态");
+			isBroken=true;
 		}else {
 			isBroken=false;
 			log.info(id + "消息：" + msg);

src/main/java/com/github/bluesbruce/ffch/handler/KeepAliveHandler.java

@@ -3,6 +3,7 @@ package com.github.bluesbruce.ffch.handler;
 import com.github.bluesbruce.ffch.data.CommandTasker;
 import com.github.bluesbruce.ffch.data.TaskDao;
 import com.github.bluesbruce.ffch.util.ExecUtil;
+import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 
 import java.io.IOException;
@@ -46,6 +47,7 @@ public class KeepAliveHandler extends Thread{
 		return false;
 	}
 	
+	@SneakyThrows
 	@Override
 	public void run() {
 		for(;stop_index==0;) {
@@ -54,7 +56,8 @@ public class KeepAliveHandler extends Thread{
 			}
 			String id=null;
 			CommandTasker task=null;
-			
+			Thread.sleep(2000);
+			//log.info("当前任务{}",queue);
 			try {
 				while(queue.peek() != null) {
 					log.info("准备重启任务："+queue);

src/main/java/com/github/bluesbruce/spring/mqttService/consumer/MqttConsumerCallBack.java

@@ -46,7 +46,7 @@ public class MqttConsumerCallBack implements MqttCallback {
             log.info(String.format("接收消息Qos : %d", message.getQos()));
             log.info(String.format("接收消息内容 : %s", new String(message.getPayload())));
             log.info(String.format("接收消息retained : %b", message.isRetained()));
-            if (topic.contains("/rtmp/live")){
+            if (topic.contains("/rtmp/live")||topic.contains("/data/DroneBMS")){
                 if (ObjectUtils.isEmpty(mqttLiveHandle)){
                     mqttLiveHandle = SpringUtil.getBean(MqttLiveHandle.class);
                 }

src/main/java/com/github/bluesbruce/spring/mqttService/send/MqttProviderConfig.java

@@ -62,7 +62,12 @@ public class MqttProviderConfig {
             //设置回调
             client.setCallback(new MqttConsumerCallBack());
             client.connect(options);
-            client.subscribe(cmdParam.getMqttTopic(),0);
+
+            String[] topics= cmdParam.getMqttTopic().split(",");
+            for (String topic :topics) {
+                client.subscribe(topic,0);
+            }
+
         } catch(MqttException e){
             e.printStackTrace();
         }

src/main/java/com/github/bluesbruce/spring/service/MqttLiveHandle.java

@@ -1,12 +1,15 @@
 package com.github.bluesbruce.spring.service;
 
 import com.alibaba.fastjson.JSONObject;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
 import org.eclipse.paho.client.mqttv3.MqttMessage;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 import org.springframework.util.ObjectUtils;
 
 @Component
+@Slf4j
 public class MqttLiveHandle {
     @Autowired
     private RtmpLiveService rtmpLiveService;
@@ -14,13 +17,19 @@ public class MqttLiveHandle {
     public void handleLive(final String topic, MqttMessage message){
         String objmsg = new String(message.getPayload());
         final JSONObject jsonObject = JSONObject.parseObject(objmsg);
+        if (topic.contains("rtmp/live")) {
             if (!ObjectUtils.isEmpty(jsonObject.get("command"))) {
                 String cmdoperat = jsonObject.get("command").toString();
                 if (cmdoperat.equals("start")) {
                     Thread thread = new Thread(new Runnable() {
                         String code = jsonObject.get("code") == null ? topic.split("/")[2] : jsonObject.get("code").toString();
+
                         public void run() {
+                            try {
                                 rtmpLiveService.pushServer(code);
+                            } catch (InterruptedException e) {
+                                e.printStackTrace();
+                            }
                         }
                     });
                     thread.start();
@@ -28,6 +37,31 @@ public class MqttLiveHandle {
                     rtmpLiveService.stopRtmp();
                 }
             }
+        }else if(topic.contains("/data/DroneBMS")){
+            log.info("获取object：",jsonObject);
+            if (!ObjectUtils.isEmpty(jsonObject.get("parm"))) {
+                log.info("获取param：",jsonObject.get("parm"));
+                JSONObject cmdoperat = jsonObject.getJSONObject("parm");
+                String boolbms = cmdoperat.getString("Power");
+                log.info("boolbms：",jsonObject.get("Power"));
+                if (boolbms.equals("on")) {
+                    Thread thread = new Thread(new Runnable() {
+                        String code = jsonObject.get("code") == null ? topic.split("/")[2] : jsonObject.get("code").toString();
+                        public void run() {
+                            try {
+                                rtmpLiveService.pushServer(code);
+                            } catch (InterruptedException e) {
+                                e.printStackTrace();
+                            }
+                        }
+                    });
+                    thread.start();
+                } else if (boolbms.equals("off")) {
+                    log.info("关闭流");
+                    rtmpLiveService.stopRtmp();
+                }
+            }
+        }
     }
 
 

src/main/java/com/github/bluesbruce/spring/service/RtmpLiveService.java

@@ -31,19 +31,24 @@ public class RtmpLiveService {
     private MqttProviderConfig mqttProviderConfig;
 
     public static final CommandManager manager = new CommandManagerImpl();
+
     /**
      * h获取通道并推送
      */
-    public void pushServer(String code){
+    public void pushServer(String code) throws InterruptedException {
         //CommandManager manager = new CommandManagerImpl();
         String reTopic = cmdParam.getMqttTopic().replace("live", "result");
         JSONObject jsonObject = new JSONObject();
         Collection<CommandTasker> infoList = manager.queryAll();
         log.info(infoList.toString());
         if (infoList.size() > 0) {
-            jsonObject.put("code", -1);
+            //manager.stopAll();
+            //manager.destory();
+            //Thread.sleep(3000);
+            /*jsonObject.put("code", -1);
             jsonObject.put("msg", "推流服务已启动，请勿重复启动");      //推流失败
-            mqttProviderConfig.publish(2,false,reTopic,jsonObject.toJSONString());
+            mqttProviderConfig.publish(2,false,reTopic,jsonObject.toJSONString());*/
+            log.info("任务运行中");
             return;
         }
         String pushUrl = "";
@@ -63,17 +68,109 @@ public class RtmpLiveService {
                 mqttProviderConfig.publish(2, false, reTopic, object.toJSONString());
                 return;
             }
+            runRtmp(pushUrl, cmdParam.getPlayUrl(), code);
             //TODO end
-        }else{
+        } else if (cmdParam.getType() == 1) {
             log.info("使用固定流媒体通道");
             pushUrl = cmdParam.getPushUrl();
-        }
             runRtmp(pushUrl, cmdParam.getPlayUrl(), code);
+        } else if (cmdParam.getType() == 3) {
+            runRtmp2to2(cmdParam.getPlayUrl());
+        } else if (cmdParam.getType() == 4) {
+            runRtmp1to2(cmdParam.getPlayUrl());
+        }
 
+
+    }
+
+    private void runRtmp2to2(String playUrl) throws InterruptedException {
+        try {
+            String reTopic = cmdParam.getMqttTopic().replace("live", "result");
+            String taskId = manager.start("test1", CommandBuidlerFactory.createBuidler()
+                    .add("ffmpeg").add("-i", playUrl)
+                    .add("-rtsp_transport", "tcp")
+                    .add("-vcodec", "copy")
+                    .add("-acodec", "copy")
+                    .add("-f", "flv")
+                    .add("-b:v", "2M")
+                    .add("-maxrate", "2M")
+                    .add("-bufsize", "1M")
+                    .add("-y").add(cmdParam.getPushUrl()));
+            Thread.sleep(3000);
+            String taskId2 = manager.start("test2", CommandBuidlerFactory.createBuidler()
+                    .add("ffmpeg").add("-i", playUrl)
+                    .add("-rtsp_transport", "tcp")
+                    .add("-vcodec", "copy")
+                    .add("-acodec", "copy")
+                    .add("-f", "flv")
+                    .add("-b:v", "2M")
+                    .add("-maxrate", "2M")
+                    .add("-bufsize", "1M")
+                    .add("-y").add(cmdParam.getPushUrl2()));
+            JSONObject jsonObject = new JSONObject();
+            if (StringUtils.isNullOrEmpty(taskId) && StringUtils.isNullOrEmpty(taskId2)) {
+                jsonObject.put("code", -1);
+                jsonObject.put("msg", "推流服务失败");      //推流失败
+                mqttProviderConfig.publish(2, false, reTopic, jsonObject.toJSONString());
+                // 停止全部任务
+                manager.stopAll();
+                return;
+            } else {
+                jsonObject.put("code", 0);
+                jsonObject.put("msg", "推流服务启动成功");
+                mqttProviderConfig.publish(2, false, reTopic, jsonObject.toJSONString());
+            }
+            Thread.sleep(cmdParam.getTime());
+            log.info(manager.queryAll().toString());
+            // 停止全部任务
+            manager.stopAll();
+            manager.destory();
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+
+    private void runRtmp1to2(String playUrl) {
+        try {
+            //单进程 多路
+            String reTopic = cmdParam.getMqttTopic().replace("live", "result");
+            String taskId = manager.start("test4", CommandBuidlerFactory.createBuidler()
+                    .add("ffmpeg").add("-i", playUrl)
+                    .add("-rtsp_transport", "tcp")
+                    .add("-vcodec", "copy")
+                    .add("-acodec", "copy")
+                    .add("-b:v", "2M")
+                    .add("-maxrate", "2M")
+                    .add("-bufsize", "1M")
+                    //多路推流测试
+                    .add("-map", "0")
+                    .add("-f").add(cmdParam.getPushUrlMap()));
+            JSONObject jsonObject = new JSONObject();
+            if (StringUtils.isNullOrEmpty(taskId)) {
+                jsonObject.put("code", -1);
+                jsonObject.put("msg", "推流服务失败");      //推流失败
+                mqttProviderConfig.publish(2, false, reTopic, jsonObject.toJSONString());
+                // 停止全部任务
+                manager.stopAll();
+                return;
+            } else {
+                jsonObject.put("code", 0);
+                jsonObject.put("msg", "推流服务启动成功");
+                mqttProviderConfig.publish(2, false, reTopic, jsonObject.toJSONString());
+            }
+            Thread.sleep(cmdParam.getTime());
+            log.info(manager.queryAll().toString());
+            // 停止全部任务
+            manager.stopAll();
+            manager.destory();
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
     }
 
     /**
      * 获取阿里云通道
+     *
      * @return
      */
     private JSONObject getChenl() {
@@ -112,6 +209,7 @@ public class RtmpLiveService {
 
     /**
      * 根据推拉流地址推动
+     *
      * @param pushUrl
      * @param playUrl
      */
@@ -125,6 +223,10 @@ public class RtmpLiveService {
                     .add("-vcodec", "copy")
                     .add("-acodec", "copy")
                     .add("-f", "flv")
+                    .add("-b:v", "2M")
+                    .add("-maxrate", "2M")
+                    .add("-bufsize", "1M")
+                    .add("-v", "trace ")
                     .add("-y").add(pushUrl));
             cmdParam.getMqttTopic();
             String reTopic = cmdParam.getMqttTopic().replace("live", "result");
@@ -143,7 +245,7 @@ public class RtmpLiveService {
                 jsonObject.put("msg", "推流服务启动成功");
                 mqttProviderConfig.publish(2, false, reTopic, jsonObject.toJSONString());
             }
-            Thread.sleep(36000);
+            Thread.sleep(cmdParam.getTime());
             log.info(manager.queryAll().toString());
             // 停止全部任务
             manager.stopAll();
@@ -185,6 +287,7 @@ public class RtmpLiveService {
 
     /**
      * 释放阿里云通道
+     *
      * @return
      */
     private JSONObject returnChenl() {