添加定时删除功能

pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-parent</artifactId>
-        <version>2.3.5.RELEASE</version>
+        <version>2.7.18</version>
         <relativePath/> <!-- lookup parent from repository -->
     </parent>
     <groupId>com.example</groupId>
@@ -13,46 +13,56 @@
     <version>0.0.1-SNAPSHOT</version>
     <name>demo</name>
     <description>demo</description>
+    <url/>
+    <licenses>
+        <license/>
+    </licenses>
+    <developers>
+        <developer/>
+    </developers>
+    <scm>
+        <connection/>
+        <developerConnection/>
+        <tag/>
+        <url/>
+    </scm>
     <properties>
         <java.version>11</java.version>
     </properties>
     <dependencies>
-
-        <dependency>
-            <groupId>org.springframework.cloud</groupId>
-            <artifactId>spring-cloud-starter-consul-discovery</artifactId>
-        </dependency>
-
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-web</artifactId>
         </dependency>
 
+
         <dependency>
-            <groupId>org.springframework.boot</groupId>
+            <groupId>io.minio</groupId>
-            <artifactId>spring-boot-starter-test</artifactId>
+            <artifactId>minio</artifactId>
-            <scope>test</scope>
+            <version>3.0.10</version>
+        </dependency>
+        <dependency>
+            <groupId>com.alibaba.fastjson2</groupId>
+            <artifactId>fastjson2</artifactId>
+            <version>2.0.53</version>
+        </dependency>
+        <!-- SkyWalking 工具包（Logback 支持） -->
+        <dependency>
+            <groupId>org.apache.skywalking</groupId>
+            <artifactId>apm-toolkit-logback-1.x</artifactId>
+            <version>8.6.0</version>
         </dependency>
 
-        <!--健康检查端点-->
         <dependency>
-            <groupId>org.springframework.boot</groupId>
+            <groupId>org.apache.skywalking</groupId>
-            <artifactId>spring-boot-starter-actuator</artifactId>
+            <artifactId>apm-toolkit-trace</artifactId>
+            <version>8.6.0</version>
         </dependency>
 
+
+
     </dependencies>
 
-    <dependencyManagement>
-        <dependencies>
-            <dependency>
-                <groupId>org.springframework.cloud</groupId>
-                <artifactId>spring-cloud-dependencies</artifactId>
-                <version>Hoxton.SR12</version>
-                <type>pom</type>
-                <scope>import</scope>
-            </dependency>
-        </dependencies>
-    </dependencyManagement>
     <build>
         <plugins>
             <plugin>
@@ -62,5 +72,4 @@
         </plugins>
     </build>
 
-
 </project>

src/main/java/com/tuoheng/steam/controller/StreamRecordController.java

@@ -1,25 +1,230 @@
 package com.tuoheng.steam.controller;
+
+
+import com.alibaba.fastjson2.JSON;
+import com.tuoheng.steam.controller.dto.PageInfo;
+import com.tuoheng.steam.controller.dto.PageStreamRequest;
+import com.tuoheng.steam.controller.dto.Response;
+import com.tuoheng.steam.dos.Mp4Info;
+import com.tuoheng.steam.dos.StreamTask;
+import com.tuoheng.steam.service.IRecordService;
+import com.tuoheng.steam.service.ITaskService;
+import com.tuoheng.steam.service.dos.DayRecord;
+import com.tuoheng.steam.service.dos.Mp4Record;
+import com.tuoheng.steam.service.dos.StreamRecord;
+import org.apache.logging.log4j.util.Strings;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.cloud.client.ServiceInstance;
+import org.springframework.beans.factory.annotation.Value;
-import org.springframework.cloud.client.discovery.DiscoveryClient;
 import org.springframework.web.bind.annotation.*;
 
-import java.util.List;
+import java.io.File;
-
+import java.util.*;
+import java.util.stream.Collectors;
 
 @RestController()
-
+@RequestMapping("/record")
 public class StreamRecordController {
 
-    @Autowired
+    private static final Logger logger = LoggerFactory.getLogger(StreamRecordController.class);
-    private DiscoveryClient discoveryClient;
 
-    @GetMapping("/")
+    @Autowired
-    public String home() {
+    ITaskService taskService;
-        List<ServiceInstance> instances =  discoveryClient.getInstances("demo");
+
-        List<String>  services = discoveryClient.getServices();
+    @Autowired
-        return "Hello World!";
+    IRecordService iRecordService;
+
+
+    @GetMapping("pic")
+    public Response<String> startPic(@RequestParam String streamUrl) {
+        logger.info("启动视频拍照 :"+streamUrl);
+        streamUrl = streamSwitch(streamUrl);
+        logger.info("启动视频拍照_ :"+streamUrl);
+        if(Objects.isNull(streamUrl)) {
+            return Response.fail(-1);
+        }
+        String outfile = taskService.startPic(streamUrl);
+        if(Objects.nonNull(outfile)) {
+            int lastSlashIndex = outfile.lastIndexOf("/");
+            String fileName = outfile.substring(lastSlashIndex + 1);
+            return Response.success(fileName);
+        }else {
+            return Response.fail(-1);
+        }
     }
 
 
+
+    @GetMapping("start")
+    public Response<StreamTask> startRecording(@RequestParam String streamUrl) {
+        logger.info("启动录制 :"+streamUrl);
+        streamUrl = streamSwitch(streamUrl);
+        logger.info("启动录制_ :"+streamUrl);
+        if(Objects.isNull(streamUrl)) {
+            return Response.fail(-1);
+        }
+
+        Response<StreamTask> response = Response.success(taskService.startTask(streamUrl));
+        logger.info("启动录制返回 :"+ JSON.toJSONString(response));
+        return response;
+    }
+
+    @GetMapping("stop")
+    public Response<StreamTask>  stopRecording(@RequestParam String streamUrl){
+
+        logger.info("关闭录制 :"+streamUrl);
+        streamUrl = streamSwitch(streamUrl);
+        logger.info("关闭录制_ :"+streamUrl);
+        if(Objects.isNull(streamUrl)) {
+            return Response.fail(-1);
+        }
+        try {
+            Response<StreamTask> response = Response.success(taskService.stopTask(streamUrl));
+            if(Objects.isNull(response.getData()) || Objects.isNull(response.getData().getOutFileName()) || response.getData().getOutFileName().isEmpty()){
+                response.setCode(500);
+            }
+            logger.info("关闭录制返回 :"+ JSON.toJSONString(response));
+            return  response;
+        }catch (Exception e){
+            return Response.fail(-1);
+        }
+    }
+
+    @GetMapping("info")
+    public Response<StreamTask>  getLastTask(@RequestParam String streamUrl){
+        logger.info("查看录制 :"+streamUrl);
+        streamUrl = streamSwitch(streamUrl);
+        logger.info("查看录制_ :"+streamUrl);
+        if(Objects.isNull(streamUrl)) {
+            return Response.fail(-1);
+        }
+        Response<StreamTask> response = Response.success(taskService.getLastTask(streamUrl));
+        logger.info("查看录制返回 :"+ JSON.toJSONString(response));
+        return  response;
+    }
+
+
+    /**
+     * 废弃
+     * @param request
+     * @return
+     */
+    @PostMapping("search")
+    public Response<PageInfo<Mp4Info>>  streamView(@RequestBody PageStreamRequest request){
+        logger.info("查看录制 search :"+JSON.toJSONString(request));
+        if(Objects.isNull(request.getPageIndex()) || Objects.isNull(request.getPageSize())
+                || request.getPageIndex() <0 || request.getPageSize() <=0 ){
+            logger.error("查看录制 search 入参错误！");
+            return Response.fail(-100);
+        }
+
+        List<Mp4Info> mp4s = searchAll(request.getStreamId());
+
+        if(!Strings.isBlank(request.getStartTime())){
+            long filerTime = Long.parseLong(request.getStartTime()) -  15 * 60 * 1000;
+            mp4s =   mp4s.stream()
+                    .filter(info -> info.getStartTime().compareTo(Long.toString(filerTime)) >= 0)
+                    .collect(Collectors.toList()); // 将结果收集到列表中
+        }
+
+        if(!Strings.isBlank(request.getEndTime())){
+            long filerTime = Long.parseLong(request.getEndTime());
+            mp4s =   mp4s.stream()
+                    .filter(info -> info.getStartTime().compareTo(Long.toString(filerTime)) <= 0)
+                    .collect(Collectors.toList()); // 将结果收集到列表中
+        }
+        /**
+         * 倒序排列
+         */
+        mp4s.sort((o1, o2) -> o2.getStartTime().compareTo(o1.getStartTime()));
+        PageInfo<Mp4Info> pageInfo  = new PageInfo<>(mp4s,request.getPageIndex(),request.getPageSize());
+        logger.info("查看录制 search 返回:"+JSON.toJSONString(request));
+        return Response.success(pageInfo);
+    }
+
+
+    public List<Mp4Info> searchAll(String streamUrl){
+
+        List<Mp4Info>  dayMp4 = new ArrayList<>();
+        List<DayRecord>  dayRecords =  iRecordService.findDaysPath();
+        for(DayRecord dayRecord : dayRecords){
+            List<StreamRecord> streamRecords = iRecordService.findInDayRecord(dayRecord);
+            for(StreamRecord stream : streamRecords){
+                if(stream.getStreamId().equals(streamUrl)){
+                    List<Mp4Record>  mp4Records =  stream.queryMp4Records();
+                    if(Objects.nonNull(mp4Records) && !mp4Records.isEmpty()){
+                        for(Mp4Record mp4Record : mp4Records){
+                            Mp4Info mp4Info = new Mp4Info();
+                            mp4Info.setStartTime(mp4Record.getStartTime());
+                            mp4Info.setUrl(dayRecord.getDay()+
+                                    File.separator + stream.getStreamId() + File.separator + mp4Record.getMp4());
+                            dayMp4.add(mp4Info);
+                        }
+                    }
+                }
+            }
+        }
+        return  dayMp4;
+    }
+
+
+    @Value("${srs.domain}")
+    private String srsdomain;
+
+    @Value("${srs.name}")
+    private String srsname;
+
+    public String streamSwitch(String source){
+
+        if(Objects.nonNull(srsdomain) && !srsdomain.isEmpty()){
+            if(Objects.nonNull(source) && !source.isEmpty()){
+               return dockerFix(source);
+            }
+        }
+
+
+        if(source.contains("stream.t-aaron.com")){
+            return source;
+        } else if (source.contains("rtmp://live.push.t-aaron.com")){
+            source =  source.replace("rtmp://live.push.t-aaron.com","http://live.play.t-aaron.com") ;
+            if(source.endsWith("_")){
+                source = source.substring(0,source.length()-1) + ".flv" + "_";
+            }else {
+                source = source + ".flv";
+            }
+            return source;
+        }else {
+            if(source.contains("https://live.play.t-aaron.com")){
+                source = source.replace("https://live.play.t-aaron.com","http://live.play.t-aaron.com");
+            }
+            return source;
+        }
+
+    }
+
+    public  String dockerFix(String url) {
+        if (url == null || url.isEmpty()) {
+            return url;
+        }
+        url = url.replaceFirst("^https?://", "rtmp://");
+        // 删除末尾的 .flv
+        url = url.replaceFirst("\\.flv$", "");
+        if(url.contains("srs-jndsj")){
+            return url;
+        }
+        // 处理 RTMP URL
+        if (url.startsWith("rtmp://")) {
+            // 先删除端口号
+            String withoutPort = url.replaceFirst("(rtmp://[^:/]+):\\d+", "$1");
+            // 替换域名为 aaa
+            String withNewDomain = withoutPort.replaceFirst("rtmp://[^/]+", "rtmp://"+srsname);
+            // 删除 .flv 后缀
+            return withNewDomain.replaceFirst("\\.flv$", "");
+        }
+
+
+
+        return url;
+    }
 }

src/main/java/com/tuoheng/steam/controller/dto/PageInfo.java

@@ -0,0 +1,71 @@
+package com.tuoheng.steam.controller.dto;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+public class PageInfo <T> implements Serializable {
+
+    public PageInfo(List<T> list,int page,int pageSize) {
+        this.total = list.size();
+        this.pageSize = pageSize;
+        this.pageIndex = page ;
+
+        // 计算分页数据
+        int fromIndex = (page - 1) * pageSize; // 起始索引
+        int toIndex = Math.min(fromIndex + pageSize, total); // 结束索引
+
+        // 检查页码是否超出范围
+        if (fromIndex >= total) {
+            // 如果页码超出范围，返回空列表
+            this.list = Collections.emptyList();
+        } else {
+            // 截取当前页的数据
+            this.list = list.subList(fromIndex, toIndex);
+        }
+    }
+
+    private List<T> list;
+    private int total;
+
+    public int getPageIndex() {
+        return pageIndex;
+    }
+
+    public void setPageIndex(int pageIndex) {
+        this.pageIndex = pageIndex;
+    }
+
+    private int pageSize;
+    private int pageIndex;
+
+    public List<T> getList() {
+        return list;
+    }
+
+    public PageInfo(int pageIndex) {
+        this.pageIndex = pageIndex;
+    }
+
+    public void setList(List<T> list) {
+        this.list = list;
+    }
+
+    public int getTotal() {
+        return total;
+    }
+
+    public void setTotal(int total) {
+        this.total = total;
+    }
+
+    public int getPageSize() {
+        return pageSize;
+    }
+
+    public void setPageSize(int pageSize) {
+        this.pageSize = pageSize;
+    }
+
+}

src/main/java/com/tuoheng/steam/controller/dto/PageStreamRequest.java

@@ -0,0 +1,53 @@
+package com.tuoheng.steam.controller.dto;
+
+import java.io.Serializable;
+
+public class PageStreamRequest implements Serializable {
+    String streamId;
+    Integer pageIndex   ;
+    Integer pageSize;
+    String startTime;
+    String endTime;
+
+    public String getStreamId() {
+        return streamId;
+    }
+
+    public void setStreamId(String streamId) {
+        this.streamId = streamId;
+    }
+
+    public Integer getPageIndex() {
+        return pageIndex;
+    }
+
+    public void setPageIndex(Integer pageIndex) {
+        this.pageIndex = pageIndex;
+    }
+
+    public Integer getPageSize() {
+        return pageSize;
+    }
+
+    public void setPageSize(Integer pageSize) {
+        this.pageSize = pageSize;
+    }
+
+    public String getStartTime() {
+        return startTime;
+    }
+
+    public void setStartTime(String startTime) {
+        this.startTime = startTime;
+    }
+
+    public String getEndTime() {
+        return endTime;
+    }
+
+    public void setEndTime(String endTime) {
+        this.endTime = endTime;
+    }
+
+
+}

src/main/java/com/tuoheng/steam/controller/dto/Response.java

@@ -0,0 +1,37 @@
+package com.tuoheng.steam.controller.dto;
+
+import java.io.Serializable;
+
+public class Response<T> implements Serializable {
+    Integer code;
+    T data;
+
+    public T getData() {
+        return data;
+    }
+
+    public Integer getCode() {
+        return code;
+    }
+
+    public void setCode(Integer code) {
+        this.code = code;
+    }
+
+    public void setData(T data) {
+        this.data = data;
+    }
+
+  public static <T> Response<T>  success(T v){
+       Response<T> r = new Response<T>();
+       r.setData(v);
+       r.code = 200;
+       return r;
+    }
+
+    public static <T> Response<T>  fail(int code){
+        Response<T> r = new Response<T>();
+        r.code = code;
+        return r;
+    }
+}

src/main/java/com/tuoheng/steam/dos/Mp4Info.java

@@ -0,0 +1,28 @@
+package com.tuoheng.steam.dos;
+
+import java.io.Serializable;
+import java.util.Date;
+
+public class Mp4Info  implements Serializable {
+
+    public String getUrl() {
+        return url;
+    }
+
+    public void setUrl(String url) {
+        this.url = url;
+    }
+
+
+
+    String url;
+    String startTime;
+
+    public String getStartTime() {
+        return startTime;
+    }
+
+    public void setStartTime(String startTime) {
+        this.startTime = startTime;
+    }
+}

src/main/java/com/tuoheng/steam/dos/ProcessType.java

@@ -0,0 +1,5 @@
+package com.tuoheng.steam.dos;
+
+public enum ProcessType {
+    RECORD,MERGE,PIC
+}

src/main/java/com/tuoheng/steam/dos/StreamProcess.java

@@ -0,0 +1,78 @@
+package com.tuoheng.steam.dos;
+
+import com.tuoheng.steam.util.TimeUtils;
+
+import java.io.Serializable;
+import java.util.Date;
+import java.util.concurrent.CompletableFuture;
+
+
+public class StreamProcess implements Serializable {
+
+    Process process;
+    Date createTime;
+    String stopTime;
+    ProcessType processType;
+    String fileName;
+
+
+
+    public String startTime(){
+        return TimeUtils.formatDateToString(createTime);
+    }
+
+    public String getStopTime() {
+        return stopTime;
+    }
+
+    public void setStopTime(String stopTime) {
+        this.stopTime = stopTime;
+    }
+
+    public CompletableFuture<Process> onExit(){
+        return process.onExit();
+    }
+
+    public void destroy(){
+        process.destroy();
+    }
+
+    public ProcessType getProcessType() {
+        return processType;
+    }
+
+    public void setProcessType(ProcessType processType) {
+        this.processType = processType;
+    }
+
+
+
+    public Long getInnerProcessId() {
+        return process.pid();
+    }
+
+    public Date getCreateTime() {
+        return createTime;
+    }
+
+    public void setCreateTime(Date createTime) {
+        this.createTime = createTime;
+    }
+
+    public StreamProcess(Process process, String fileName, ProcessType processType) {
+        this.process = process;
+        this.fileName = fileName;
+        this.createTime = new Date();
+        this.processType = processType;
+    }
+
+
+
+    public String getFileName() {
+        return fileName;
+    }
+
+    public void setFileName(String fileName) {
+        this.fileName = fileName;
+    }
+}

src/main/java/com/tuoheng/steam/dos/StreamTask.java

@@ -0,0 +1,136 @@
+package com.tuoheng.steam.dos;
+
+import com.tuoheng.steam.util.TimeUtils;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+
+public class StreamTask implements Serializable {
+
+    Long taskId;
+    Date startTime;
+    String strStartTime;
+
+    public String getDuration() {
+        return duration;
+    }
+
+    public void setDuration(String duration) {
+        this.duration = duration;
+    }
+
+    String duration;
+    /**
+     * 接收到结束命令的时间
+     */
+    String stopCommandTime;
+    /**
+     * 执行生成TS文件的命令
+     */
+    List<StreamProcess> streamProcesses = new ArrayList<>();
+    /**
+     * 合并进程的命令
+     */
+    StreamProcess mergeProcess;
+
+    private  static Long pid = 0L;
+
+    private synchronized static Long generatePid(){
+        if(pid==Long.MAX_VALUE){
+            pid = 0L;
+        }else {
+            pid++;
+        }
+        return pid;
+    }
+
+
+    public String getStrStartTime() {
+        return strStartTime;
+    }
+
+    public void setStrStartTime(String strStartTime) {
+        this.strStartTime = strStartTime;
+    }
+
+    public void setStartTime(String startTime) {
+        this.strStartTime = startTime;
+    }
+
+
+    public String getStopCommandTime() {
+        return stopCommandTime;
+    }
+
+    public void setStopCommandTime(String stopCommandTime) {
+        this.stopCommandTime = stopCommandTime;
+    }
+
+
+
+    public List<StreamProcess> getStreamProcesses() {
+        return streamProcesses;
+    }
+
+    public void setStreamProcesses(List<StreamProcess> streamProcesses) {
+        this.streamProcesses = streamProcesses;
+    }
+
+    public StreamProcess getMergeProcess() {
+        return mergeProcess;
+    }
+
+    public void setMergeProcess(StreamProcess mergeProcess) {
+        this.mergeProcess = mergeProcess;
+    }
+
+    public String getStreamUrl() {
+        return streamUrl;
+    }
+
+    public void setStreamUrl(String streamUrl) {
+        this.streamUrl = streamUrl;
+    }
+
+
+    String streamUrl;
+
+
+    public Long getTaskId() {
+        return taskId;
+    }
+
+    public void setTaskId(Long taskId) {
+        this.taskId = taskId;
+    }
+
+    public Date getStartTime() {
+        return startTime;
+    }
+
+    public void setStartTime(Date startTime) {
+        this.startTime = startTime;
+    }
+
+    public StreamTask(String streamUrl) {
+        this.taskId = generatePid();
+        this.streamUrl = streamUrl;
+        this.startTime = new Date();
+        this.strStartTime = TimeUtils.formatDateToString(this.startTime);
+        this.stopCommandTime = null;
+    }
+
+
+
+    public String getOutFileName() {
+        return outFileName;
+    }
+
+    public void setOutFileName(String outFileName) {
+        this.outFileName = outFileName;
+    }
+
+    String outFileName;
+}

src/main/java/com/tuoheng/steam/schedule/Scheduler.java

@@ -0,0 +1,78 @@
+package com.tuoheng.steam.schedule;
+
+import com.tuoheng.steam.service.IRecordService;
+import com.tuoheng.steam.service.TaskService;
+import com.tuoheng.steam.service.dos.DayRecord;
+import com.tuoheng.steam.service.dos.FlvRecord;
+import com.tuoheng.steam.service.dos.Mp4Record;
+import com.tuoheng.steam.service.dos.StreamRecord;
+import com.tuoheng.steam.util.TimeUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import java.util.*;
+
+
+@Component
+public class Scheduler {
+
+    private static final Logger logger = LoggerFactory.getLogger(Scheduler.class);
+
+    @Autowired
+    IRecordService iRecordService;
+
+    @Value("${livedates}")
+    private Integer livedates;
+
+    @Value("${cangneiwai}")
+    private Boolean cangneiwai;
+    /**
+     * 初次执行延迟6秒执行
+     * 每隔 60 分钟执行一次 60*60*1000
+     */
+    @Scheduled(fixedRate = 3600000, initialDelay = 6000)
+    public void mergeTask() {
+
+        if(!cangneiwai){
+            logger.info("舱内外无需录制");
+            return;
+        }
+
+        logger.info("开始FLV到MP4的转换 - " + System.currentTimeMillis() / 1000);
+        List<DayRecord>  dayRecords = iRecordService.findDaysPath();
+        for (int index = 0; index < dayRecords.size(); index++) {
+
+            DayRecord dayRecord = dayRecords.get(index);
+            if(TimeUtils.isBefore(dayRecord.getDay(),livedates)){
+                dayRecord.clear();
+            } else {
+                List<StreamRecord> streamRecords = dayRecord.getStreamRecords();
+                for(StreamRecord streamRecord : streamRecords){
+                    List<FlvRecord>  flvRecords = streamRecord.queryFlvRecords();
+                    for(FlvRecord flvRecord : flvRecords){
+                        iRecordService.mergeMp4(flvRecord);
+                    }
+                }
+            }
+        }
+    }
+
+
+
+
+//    public static boolean isWithin15Minutes(String timestamp1, String timestamp2) {
+//        // 将字符串转换为 long 类型
+//        long time1 = Long.parseLong(timestamp1);
+//        long time2 = Long.parseLong(timestamp2);
+//        // 计算时间差的绝对值
+//        long diff = Math.abs(time2 - time1);
+//        // 15 分钟 = 15 * 60 * 1000 毫秒
+//        long fifteenMinutesInMillis = 15 * 60 * 1000;
+//        // 判断是否小于或等于 15 分钟
+//        return diff <= fifteenMinutesInMillis;
+//    }
+
+}

src/main/java/com/tuoheng/steam/service/IRecordService.java

@@ -0,0 +1,19 @@
+package com.tuoheng.steam.service;
+
+import com.tuoheng.steam.service.dos.*;
+
+import java.util.Date;
+import java.util.List;
+
+/**
+ * 视频回放服务
+ */
+public interface IRecordService {
+
+    public List<DayRecord> findDaysPath();
+
+    public List<StreamRecord> findInDayRecord(DayRecord dayRecord);
+
+    public void mergeMp4(FlvRecord flvRecord);
+
+}

src/main/java/com/tuoheng/steam/service/ITaskService.java

@@ -0,0 +1,19 @@
+package com.tuoheng.steam.service;
+
+import com.tuoheng.steam.dos.StreamTask;
+
+/**
+ * 视频录制服务
+ */
+public interface ITaskService {
+
+    public StreamTask startTask(String streamUrl);
+
+    public String startPic(String streamUrl);
+
+    public StreamTask stopTask(String streamUrl) throws Exception;
+
+    public StreamTask getLastTask(String streamUrl);
+
+    public void stopAllTask();
+}

src/main/java/com/tuoheng/steam/service/RecordService.java

@@ -0,0 +1,67 @@
+package com.tuoheng.steam.service;
+
+import com.tuoheng.steam.service.dos.*;
+import com.tuoheng.steam.service.innerService.ProcessService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+import java.io.File;
+import java.util.*;
+import java.util.stream.Collectors;
+
+import static com.tuoheng.steam.util.FileUtil.readFiles;
+
+@Component
+public class RecordService implements IRecordService {
+
+    @Value("${recordPath}")
+    private String recordPath;
+
+    @Autowired
+    ProcessService processService;
+
+    private static final Logger logger = LoggerFactory.getLogger(IRecordService.class);
+
+    public static boolean isValidDate(String dateStr) {
+        // 正则表达式：4位年份 + 2位月份 + 2位日期
+        String regex = "^\\d{4}(0[1-9]|1[0-2])(0[1-9]|[12][0-9]|3[01])$";
+        return dateStr.matches(regex);
+    }
+
+    @Override
+    public List<DayRecord> findDaysPath() {
+        List<DayRecord> days = new ArrayList<>();
+        List<String> fileNames = readFiles(recordPath);
+        for (String fileName : fileNames) {
+            if(isValidDate(fileName)){
+                DayRecord day = new DayRecord();
+                day.setRoot(recordPath);
+                day.setDay(fileName);
+                days.add(day);
+            }
+        }
+        days.sort(Comparator.comparing(DayRecord::getDay));
+        return days;
+    }
+
+
+    @Override
+    public List<StreamRecord> findInDayRecord(DayRecord dayRecord) {
+        return   dayRecord.getStreamRecords();
+    }
+
+    @Override
+    public void mergeMp4(FlvRecord flvRecord) {
+        try {
+            processService.mergeMp4(flvRecord);
+        }catch (Exception e){
+            logger.error(e.getMessage());
+        }
+    }
+
+
+
+}

src/main/java/com/tuoheng/steam/service/TaskService.java

@@ -0,0 +1,367 @@
+package com.tuoheng.steam.service;
+
+import com.alibaba.fastjson2.JSON;
+import com.tuoheng.steam.dos.StreamProcess;
+import com.tuoheng.steam.dos.StreamTask;
+import com.tuoheng.steam.service.innerService.ProcessService;
+import com.tuoheng.steam.util.FileUtil;
+import com.tuoheng.steam.util.ProcessManager;
+import com.tuoheng.steam.util.TimeUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.scheduling.annotation.EnableScheduling;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Service;
+import org.springframework.util.CollectionUtils;
+
+import javax.annotation.PostConstruct;
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.*;
+import java.util.concurrent.*;
+
+@Service
+@EnableScheduling
+public class TaskService implements ITaskService{
+
+    @Value("${srs.targetPath}")
+    private String targetPath;
+
+//    @Value("${ffmpeg}")
+    private String ffmpeg ="ffmpeg";
+
+    ScheduledExecutorService executor = Executors.newScheduledThreadPool(4);
+
+    /**
+     * 清除没有被关闭的进程
+     */
+    @Scheduled(fixedRate = 60000)
+    public void cleaTask() {
+        for (ConcurrentHashMap.Entry<String, StreamTask> entry : runningTasks.entrySet()) {
+            StreamTask streamTask = entry.getValue();
+            if(streamTask.getStartTime().getTime() < new Date().getTime() -  90 * 60 * 1000 ) {
+                logger.info("清理废弃任务 {}", JSON.toJSONString(streamTask));
+                try {
+                    StreamTask s = stopTask(entry.getKey());
+                    scheduler.schedule(() -> {
+                        File file = new File(targetPath +"/"+ s.getOutFileName());
+                        try {
+                            logger.info("废弃文件删除成功 {}",file.delete());
+                        }catch (Exception e) {
+                            logger.error(e.getMessage());
+                        }
+                    }, 60, TimeUnit.SECONDS);
+                }catch (Exception ignore) {}
+
+            }
+        }
+    }
+
+    private static final Logger logger = LoggerFactory.getLogger(TaskService.class);
+
+    @Autowired
+    ProcessService processService;
+
+    /**
+     * 执行中的任务
+     */
+    ConcurrentHashMap<String, StreamTask> runningTasks = new ConcurrentHashMap<>();
+
+    /**
+     * 历史任务
+     */
+    ConcurrentHashMap<String, Deque<StreamTask>> historyTasks = new ConcurrentHashMap<>();
+
+    /**
+     * 任务池
+     */
+    ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
+
+    @PostConstruct
+    public void init() {
+
+        /**
+         *
+         */
+        scheduler.scheduleWithFixedDelay(new Runnable() {
+            @Override
+            public void run() {
+                runningTasks.forEach((key, value) -> {
+                    Calendar calendar = Calendar.getInstance();
+                    calendar.add(Calendar.HOUR, -2);
+                    Date twoHoursAgo = calendar.getTime();
+                    if (value.getStartTime().before(twoHoursAgo)) {
+                        logger.error("taskId {} 执行超时,手动关闭", key);
+                        try {
+                            stopTask(key);
+                        }catch (Exception ignore) {}
+                    }
+                });
+
+                for (Map.Entry<String, Deque<StreamTask>> entry : historyTasks.entrySet()) {
+
+                    for (StreamTask task : entry.getValue()) {
+                        List<StreamProcess> streamProcesses = task.getStreamProcesses();
+                        for (StreamProcess process : streamProcesses) {
+                            process.destroy();
+                            File file = new File(process.getFileName());
+                            FileUtil.deleteFile(file);
+
+                        }
+                    }
+
+                    while (entry.getValue().size()>10){
+                        entry.getValue().pollFirst();
+                    }
+                }
+            }
+        }, 60, 60, TimeUnit.SECONDS);
+    }
+
+    public String startPic(String streamUrl) {
+
+        try {
+            StreamProcess streamProcess = processService.takePic(streamUrl);
+            String outFileName = streamProcess.getFileName();
+            /*
+              注册进程
+             */
+            ProcessManager.registerProcess(streamProcess.getInnerProcessId());
+            logger.info("streamUrl  {} startPicProcess {} ", streamUrl,streamProcess.getInnerProcessId());
+            CompletableFuture<Process> future = streamProcess.onExit();
+            future.thenRun(() -> {
+                logger.info("streamUrl  {} startPicProcess {} Over ", streamUrl,streamProcess.getInnerProcessId());
+                streamProcess.setStopTime(TimeUtils.formatDateToString(new Date()));
+                ProcessManager.unRegisterProcess(streamProcess.getInnerProcessId());
+            }).exceptionally(ex -> {
+                logger.info("streamUrl  {} startPicProcess {} Exceptionally ", streamUrl,streamProcess.getInnerProcessId());
+                streamProcess.setStopTime(TimeUtils.formatDateToString(new Date()));
+                ProcessManager.unRegisterProcess(streamProcess.getInnerProcessId());
+                return null;
+            });
+            return outFileName;
+        }catch (Exception e) {
+            logger.info("streamUrl  {}  ", streamUrl,e);
+        }
+        return null;
+    }
+
+    public StreamTask startTask(String streamUrl) {
+
+        /*
+          当前有任务,则返回当前的任务
+         */
+        if (runningTasks.containsKey(streamUrl)) {
+            return runningTasks.get(streamUrl);
+        }
+
+        /*
+          当前无任务,则开启新任务
+         */
+        StreamTask taskInstance = new StreamTask(streamUrl);
+        if (runningTasks.putIfAbsent(streamUrl, taskInstance) == null) {
+            logger.info("streamUrl {} startTask {} ", streamUrl, taskInstance.getTaskId());
+            startTask(streamUrl, taskInstance);
+         //   runningTasks.put(streamUrl, taskInstance);
+            return taskInstance;
+        } else {
+            return runningTasks.get(streamUrl);
+        }
+
+    }
+
+    public StreamTask stopTask(String streamUrl) throws Exception {
+
+        boolean recordSuccess = true;
+        StreamTask currentStreamTask = runningTasks.remove(streamUrl);
+
+        if (currentStreamTask!= null) {
+            currentStreamTask.setStopCommandTime(TimeUtils.formatDateToString(new Date()));
+
+            if (!CollectionUtils.isEmpty(currentStreamTask.getStreamProcesses())) {
+                for (StreamProcess streamProcess : currentStreamTask.getStreamProcesses()) {
+                    if(!new File(streamProcess.getFileName()).exists()){
+                        recordSuccess = false;
+                        logger.error("流录制失败: streamUrl {} taskId {} destroy Process {}", streamUrl,currentStreamTask.getTaskId() ,streamProcess.getInnerProcessId());
+                    }else {
+                        logger.info("streamUrl {} taskId {} destroy Process {}", streamUrl,currentStreamTask.getTaskId() ,streamProcess.getInnerProcessId());
+                    }
+                    streamProcess.destroy();
+                }
+            }
+
+            if(!recordSuccess){
+                throw new Exception("");
+            }
+
+            try {
+                String outFileName = UUID.randomUUID().toString() + ".mp4";
+                StreamProcess mergeProcess = processService.mergeStream(outFileName, currentStreamTask.getStreamProcesses());
+
+                if(Objects.nonNull(mergeProcess)){
+                    /**
+                     * 注册进程
+                     */
+                    ProcessManager.registerProcess(mergeProcess.getInnerProcessId());
+                    currentStreamTask.setMergeProcess(mergeProcess);
+
+
+                    logger.info("streamUrl {} taskId {} startMergeProcess {} ", streamUrl, currentStreamTask.getTaskId(),
+                            mergeProcess.getInnerProcessId());
+
+                    CompletableFuture<Process> future = mergeProcess.onExit();
+                    future.thenRun(() -> {
+                        logger.info("streamUrl {} taskId {} MergeProcess {} Over", streamUrl,currentStreamTask.getTaskId(), mergeProcess.getInnerProcessId());
+                        mergeProcess.setStopTime(TimeUtils.formatDateToString(new Date()));
+                        ProcessManager.unRegisterProcess(mergeProcess.getInnerProcessId());
+                        for(StreamProcess streamProcess :currentStreamTask.getStreamProcesses()){
+                            File file = new File(streamProcess.getFileName());
+                            FileUtil.deleteFile(file);
+                        }
+                    }).exceptionally(ex -> {
+                        logger.error("streamUrl {} taskId {} MergeProcess {} exceptionally", streamUrl,currentStreamTask.getTaskId(), mergeProcess.getInnerProcessId(), ex);
+                        mergeProcess.setStopTime(TimeUtils.formatDateToString(new Date()));
+                        ProcessManager.unRegisterProcess(mergeProcess.getInnerProcessId());
+                        for(StreamProcess streamProcess :currentStreamTask.getStreamProcesses()){
+                            File file = new File(streamProcess.getFileName());
+                            FileUtil.deleteFile(file);
+                        }
+                        return null;
+                    });
+                    currentStreamTask.setOutFileName(outFileName);
+                }
+
+
+                if(Objects.nonNull(currentStreamTask.getMergeProcess())){
+                    if(Objects.nonNull(currentStreamTask.getMergeProcess().getFileName()) &&
+                            !currentStreamTask.getMergeProcess().getFileName().isEmpty()){
+                        try {
+                            executor.schedule(new Runnable() {
+                                @Override
+                                public void run() {
+                                    try {
+                                        logger.info("FileName :"+currentStreamTask.getMergeProcess().getFileName());
+                                        Process process = Runtime.getRuntime().exec(ffmpeg + " -i " +
+                                                currentStreamTask.getMergeProcess().getFileName());
+                                        BufferedReader reader = new BufferedReader(new InputStreamReader(process.getErrorStream()));
+                                        String line;
+                                        while ((line = reader.readLine()) != null) {
+                                            logger.info("Duration line: " + line);
+                                            if (line.contains("Duration")) {
+                                                String duration = line.split("Duration: ")[1].split(",")[0];
+                                                logger.info("Duration: " + duration.substring(0, 8));
+                                                currentStreamTask.setDuration(duration.substring(0, 8));
+                                                logger.info("-----------------放入缓存-----------------");
+                                                /**
+                                                 * 放入缓存
+                                                 */
+                                                if( historyTasks.containsKey(streamUrl)){
+                                                    historyTasks.get(streamUrl).add(currentStreamTask);
+                                                }else {
+                                                    historyTasks.put(streamUrl, new LinkedList<>());
+                                                    historyTasks.get(streamUrl).offerLast(currentStreamTask);
+                                                }
+                                                break;
+                                            }
+                                        }
+                                        reader.close();
+                                    }catch (Exception e){
+                                        logger.info(e.getMessage());
+                                    }
+                                }
+                            },40, TimeUnit.SECONDS);
+
+                        } catch (Exception e) {
+                            e.printStackTrace();
+                        }
+                    }
+
+                }
+
+                return currentStreamTask;
+            } catch (Exception e) {
+                logger.info("taskId {} Stop exceptionally", streamUrl,e);
+                return null;
+            }
+
+        }else {
+            throw new Exception("");
+//            if(historyTasks.containsKey(streamUrl)){
+//                return historyTasks.get(streamUrl).peekLast();
+//            }else {
+//                return null;
+//            }
+        }
+    }
+
+    public StreamTask getLastTask(String streamUrl) {
+        StreamTask current =  runningTasks.get(streamUrl);
+        if(Objects.nonNull(current)){
+            return current;
+        }else {
+            if(historyTasks.containsKey(streamUrl)){
+                return historyTasks.get(streamUrl).peekLast();
+            }
+            return null;
+        }
+    }
+
+    private void startTask(String streamUrl, StreamTask streamTask) {
+
+        try {
+            /*
+              任务不存在了,直接退出
+             */
+            if (streamTask.getStopCommandTime() != null) {
+                logger.info("streamUrl {} Task {} Stopped", streamUrl, streamTask.getTaskId());
+                return;
+            }
+
+            StreamProcess streamProcess = processService.recordStream(streamUrl);
+            /**
+             * 注册进程
+             */
+            ProcessManager.registerProcess(streamProcess.getInnerProcessId());
+            streamTask.getStreamProcesses().add(streamProcess);
+
+            logger.info("streamUrl {} taskId {} startProcess {} ", streamUrl, streamTask.getTaskId(),streamProcess.getInnerProcessId());
+
+            CompletableFuture<Process> future = streamProcess.onExit();
+            future.thenRun(() -> {
+                logger.info("streamUrl {} taskId {} Process {} Over", streamUrl, streamTask.getTaskId(),streamProcess.getInnerProcessId());
+                streamProcess.setStopTime(TimeUtils.formatDateToString(new Date()));
+                ProcessManager.unRegisterProcess(streamProcess.getInnerProcessId());
+                try {
+                    Thread.sleep(2000L);
+                } catch (Exception ignored){}
+                if(!new File(streamProcess.getFileName()).exists()){
+                    streamTask.getStreamProcesses().remove(streamProcess);
+                }
+                startTask(streamUrl, streamTask);
+            }).exceptionally(ex -> {
+                logger.info("streamUrl {} taskId {} Process {} Exceptionally", streamUrl, streamTask.getTaskId(),streamProcess.getInnerProcessId(),ex);
+                streamProcess.setStopTime(TimeUtils.formatDateToString(new Date()));
+                startTask(streamUrl, streamTask);
+                ProcessManager.unRegisterProcess(streamProcess.getInnerProcessId());
+                return null;
+            });
+
+        } catch (IOException e) {
+            logger.info("streamUrl {} taskId {} startProcess Exception", streamUrl,streamTask.getTaskId(),e);
+        }
+
+    }
+
+
+    public void stopAllTask(){
+        runningTasks.forEach((key, value) -> {
+            try {
+                stopTask(key);
+            }catch (Exception ignore){}
+        });
+    }
+}

src/main/java/com/tuoheng/steam/service/dos/DayRecord.java

@@ -0,0 +1,50 @@
+package com.tuoheng.steam.service.dos;
+
+import com.tuoheng.steam.util.FileUtil;
+
+import java.io.File;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * 代表日期的文件夹
+ */
+public class DayRecord {
+    public String getRoot() {
+        return root;
+    }
+
+    public void setRoot(String root) {
+        this.root = root;
+    }
+
+    String root;
+
+    String day;
+
+    public String getDay() {
+        return day;
+    }
+
+    public void setDay(String day) {
+        this.day = day;
+    }
+
+    public void clear(){
+        FileUtil.deleteFolder(root + File.separator + this.day);
+    }
+
+    public List<StreamRecord> getStreamRecords() {
+        List<StreamRecord> result = new ArrayList<>();
+        List<String>  fileList = FileUtil.readFiles(root + File.separator + day);
+        for(String fileName : fileList){
+            StreamRecord record = new StreamRecord();
+            record.setDayRecord(this);
+            record.setStreamId(fileName);
+            record.setStreamType(FileUtil.getStreamType(fileName));
+            result.add(record);
+        }
+        return result;
+    }
+}

src/main/java/com/tuoheng/steam/service/dos/FlvRecord.java

@@ -0,0 +1,44 @@
+package com.tuoheng.steam.service.dos;
+
+/**
+ * FLV文件
+ */
+public class FlvRecord {
+    public String getFlv() {
+        return flv;
+    }
+
+    public void setFlv(String flv) {
+        this.flv = flv;
+    }
+
+    public StreamRecord getStream() {
+        return stream;
+    }
+
+    public void setStream(StreamRecord stream) {
+        this.stream = stream;
+    }
+
+    String flv;
+    StreamRecord stream;
+
+    public String getStartTime() {
+        return extractNameWithoutExtension(flv);
+    }
+
+
+    public static String extractNameWithoutExtension(String fileName) {
+        // 找到最后一个 '.' 的位置
+        int lastDotIndex = fileName.lastIndexOf('.');
+
+        // 如果文件名中没有 '.'，则返回整个文件名
+        if (lastDotIndex == -1) {
+            return fileName;
+        }
+
+        // 截取从开头到最后一个 '.' 之前的部分
+        return fileName.substring(0, lastDotIndex);
+    }
+
+}

src/main/java/com/tuoheng/steam/service/dos/Mp4Record.java

@@ -0,0 +1,45 @@
+package com.tuoheng.steam.service.dos;
+
+/**
+ * MP4文件
+ */
+public class Mp4Record {
+
+    String mp4;
+    StreamRecord stream;
+
+
+
+    public String getMp4() {
+        return mp4;
+    }
+
+    public void setMp4(String mp4) {
+        this.mp4 = mp4;
+    }
+
+    public StreamRecord getStream() {
+        return stream;
+    }
+
+    public void setStream(StreamRecord stream) {
+        this.stream = stream;
+    }
+
+    public String getStartTime() {
+        return extractNameWithoutExtension(mp4);
+    }
+
+    public static String extractNameWithoutExtension(String fileName) {
+        // 找到最后一个 '.' 的位置
+        int lastDotIndex = fileName.lastIndexOf('.');
+
+        // 如果文件名中没有 '.'，则返回整个文件名
+        if (lastDotIndex == -1) {
+            return fileName;
+        }
+
+        // 截取从开头到最后一个 '.' 之前的部分
+        return fileName.substring(0, lastDotIndex);
+    }
+}

src/main/java/com/tuoheng/steam/service/dos/StreamRecord.java

@@ -0,0 +1,95 @@
+package com.tuoheng.steam.service.dos;
+
+import com.tuoheng.steam.util.FileUtil;
+
+import java.io.File;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+
+/**
+ * 代表流的文件夹
+ */
+public class StreamRecord {
+    public String getStreamId() {
+        return streamId;
+    }
+
+    public void setStreamId(String streamId) {
+        this.streamId = streamId;
+    }
+
+    public DayRecord getDayRecord() {
+        return dayRecord;
+    }
+
+    public void setDayRecord(DayRecord dayRecord) {
+        this.dayRecord = dayRecord;
+    }
+
+    public StreamType getStreamType() {
+        return streamType;
+    }
+
+    public void setStreamType(StreamType streamType) {
+        this.streamType = streamType;
+    }
+
+    String streamId;
+    StreamType streamType;
+    DayRecord dayRecord;
+
+    public List<FlvRecord> queryFlvRecords(){
+
+        List<FlvRecord> result = new ArrayList<>();
+        List<String> fileList = FileUtil.readFiles(
+                dayRecord.root+  File.separator + dayRecord.day + File.separator + streamId);
+
+        for(String fileName : fileList){
+            if(FileUtil.isFlv(fileName)){
+                FlvRecord flvRecord = new FlvRecord();
+                flvRecord.setFlv(fileName);
+
+                flvRecord.setStream(this);
+                result.add(flvRecord);
+            }
+        }
+
+        result.sort((o1, o2) -> {
+            // 提取时间戳并转换为 long 类型
+            long timestamp1 = Long.parseLong(o1.getFlv().split("\\.")[0]);
+            long timestamp2 = Long.parseLong(o2.getFlv().split("\\.")[0]);
+            // 按时间戳升序排序
+            return Long.compare(timestamp1, timestamp2);
+        });
+
+        return result;
+    }
+
+    public List<Mp4Record> queryMp4Records(){
+
+        List<Mp4Record> result = new ArrayList<>();
+        List<String> fileList = FileUtil.readFiles(
+                dayRecord.root+  File.separator + dayRecord.day + File.separator + streamId);
+        for(String fileName : fileList){
+            if(FileUtil.isMp4(fileName)){
+                Mp4Record record = new Mp4Record();
+                record.setStream(this);
+                record.setMp4(fileName);
+                result.add(record);
+            }
+        }
+
+        result.sort((o1, o2) -> {
+            // 提取时间戳并转换为 long 类型
+            long timestamp1 = Long.parseLong(o1.mp4.split("\\.")[0]);
+            long timestamp2 = Long.parseLong(o2.mp4.split("\\.")[0]);
+            // 按时间戳升序排序
+            return Long.compare(timestamp1, timestamp2);
+        });
+
+        return result;
+    }
+}

src/main/java/com/tuoheng/steam/service/dos/StreamType.java

@@ -0,0 +1,5 @@
+package com.tuoheng.steam.service.dos;
+
+public enum StreamType {
+    Inner,Outer,LiveVideo
+}

src/main/java/com/tuoheng/steam/service/innerService/ProcessService.java

@@ -0,0 +1,249 @@
+package com.tuoheng.steam.service.innerService;
+
+import com.tuoheng.steam.dos.StreamProcess;
+import com.tuoheng.steam.dos.ProcessType;
+import com.tuoheng.steam.service.dos.FlvRecord;
+import com.tuoheng.steam.util.FileUtil;
+import com.tuoheng.steam.util.ProcessManager;
+import com.tuoheng.steam.util.TimeUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.stream.Collectors;
+
+@Service
+public class ProcessService {
+
+    @Value("${srs.splitPath}")
+    private String splitPath;
+
+//    @Value("${ffmpeg}")
+    private String ffmpeg = "ffmpeg";
+
+    @Value("${srs.targetPath}")
+    private String targetPath;
+
+    @Value("${recordPath}")
+    private String recordPath;
+
+    private static final Logger logger = LoggerFactory.getLogger(ProcessService.class);
+
+    /**
+     * 任务池
+     */
+    ExecutorService loggingService = Executors.newCachedThreadPool();
+
+    public void mergeMp4(FlvRecord flvRecord) throws IOException, ExecutionException, InterruptedException {
+
+        String command = ffmpeg + " -i "+ recordPath+   File.separator +
+                    flvRecord.getStream().getDayRecord().getDay() +  File.separator +
+                    flvRecord.getStream().getStreamId()  +  File.separator + flvRecord.getFlv() + " -c:v copy -c:a aac " + recordPath+   File.separator +
+                    flvRecord.getStream().getDayRecord().getDay() +  File.separator +
+                    flvRecord.getStream().getStreamId()  +  File.separator +flvRecord.getStartTime() + ".mp4";
+
+
+        ProcessBuilder pb = new ProcessBuilder(command.split(" "));
+        pb.redirectErrorStream(true);
+        Process process = pb.start();
+
+        loggingService.execute(new Runnable() {
+            @Override
+            public void run() {
+                logger.info("mergeMp4 process Start {}",command);
+                try (BufferedReader reader = new BufferedReader(
+                        new InputStreamReader(process.getInputStream()))) {
+                    String line;
+                    while ((line = reader.readLine()) != null) {
+                        logger.info("mergeMp4-------- {}",line);
+                    }
+                } catch (IOException e) {
+                    logger.info("mergeMp4-------- Over",e);
+                }
+                logger.info("mergeMp4 Over");
+                File delete = new File(recordPath+   File.separator +
+                        flvRecord.getStream().getDayRecord().getDay() +  File.separator +
+                        flvRecord.getStream().getStreamId()  +  File.separator + flvRecord.getFlv());
+                FileUtil.deleteFile(delete);
+            }
+        });
+
+        CompletableFuture<Process> future = process.onExit();
+        // 阻塞等待进程结束
+        Process completedProcess = null;
+        try {
+             completedProcess = future.get(4, TimeUnit.SECONDS);
+             logger.info("mergeMp4正常完成-------- Over");
+        }catch (Exception e) {
+            String fileName =  recordPath+   File.separator +
+                    flvRecord.getStream().getDayRecord().getDay() +  File.separator +
+                    flvRecord.getStream().getStreamId()  +  File.separator +flvRecord.getStartTime() + ".mp4";
+            File file = new File(fileName);
+            if(file.exists()){
+                logger.info("mergeMp4超时4S完成-------- Over",e);
+                process.destroy();
+            } else {
+                try {
+                    completedProcess = future.get(4, TimeUnit.SECONDS);
+                }catch (Exception e1) {
+                    if(file.exists()){
+                        logger.info("mergeMp4超时8S完成-------- Over",e);
+                        process.destroy();
+                    }else {
+                        logger.info("mergeMp4超时8S未完成-------- Over",e);
+                        process.destroy();
+                    }
+                }
+            }
+        }
+
+        if(Objects.nonNull(completedProcess)) {
+            // 检查进程是否成功结束
+            if (completedProcess.exitValue() == 0) {
+                logger.info("进程成功结束！");
+            } else {
+                logger.info("进程失败，退出码 {} " ,completedProcess.exitValue());
+            }
+        }
+
+    }
+
+
+    public StreamProcess takePic(String streamUrl) throws IOException {
+        String outFileName = targetPath + "/" +UUID.randomUUID().toString() +".jpg";
+        String command = String.format(
+                ffmpeg+ " -i %s  -vf fps=1 -frames:v 1 -q:v 2 %s",
+                streamUrl, outFileName);
+
+
+
+        List<String> listCommand = new ArrayList<>();
+        listCommand.add(ffmpeg); // ffmpeg 路径
+        listCommand.add("-i");
+        listCommand.add(streamUrl); // 流地址
+        listCommand.add("-vf");
+        listCommand.add("fps=1");
+        listCommand.add("-frames:v");
+        listCommand.add("1");
+        listCommand.add("-q:v");
+        listCommand.add("2");
+        listCommand.add(outFileName); // 输出文件名
+        logger.info("takePic {}", listCommand);
+
+        ProcessBuilder pb = new ProcessBuilder(listCommand);
+        pb.redirectErrorStream(true);
+        Process process = pb.start();
+
+        loggingService.execute(new Runnable() {
+            @Override
+            public void run() {
+                try (BufferedReader reader = new BufferedReader(
+                        new InputStreamReader(process.getInputStream()))) {
+                    String line;
+                    while ((line = reader.readLine()) != null) {
+                        logger.info("recordStream-------- {}",line);
+                    }
+                } catch (IOException e) {
+                    logger.info("takePic-------- Over");
+                }
+            }
+        });
+        return new StreamProcess(process,outFileName, ProcessType.PIC);
+    }
+
+    public StreamProcess recordStream(String streamUrl) throws IOException {
+
+        String  recordFileName = splitPath+'/'+UUID.randomUUID().toString() +".ts";
+        /**
+         * 如果streamUrl末尾有个_,在实际录制的时候需要将_去除
+         */
+        String command = String.format(
+                ffmpeg+ " -i %s -c copy -f mpegts -reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 2 %s",
+                Optional.of(streamUrl)
+                        .filter(s -> s.endsWith("_"))
+                        .map(s -> s.substring(0, s.length() - 1))
+                        .orElse(streamUrl), recordFileName);
+        logger.info("recordStream {}", command);
+        ProcessBuilder pb = new ProcessBuilder(command.split(" "));
+        pb.redirectErrorStream(true);
+        Process process = pb.start();
+
+        loggingService.execute(new Runnable() {
+            @Override
+            public void run() {
+                try (BufferedReader reader = new BufferedReader(
+                        new InputStreamReader(process.getInputStream()))) {
+                    String line;
+                    while ((line = reader.readLine()) != null) {
+//                        logger.info("recordStream-------- {}",line);
+                    }
+                } catch (IOException e) {
+                    logger.info("recordStream-------- Over");
+                }
+            }
+        });
+
+        return new StreamProcess(process,recordFileName, ProcessType.RECORD);
+
+    };
+
+    public StreamProcess mergeStream(String outFileName, List<StreamProcess> streamProcesses) throws IOException {
+
+        List<String> fileList = new ArrayList<>();
+        for(StreamProcess streamProcess : streamProcesses){
+            File newFile = new File(streamProcess.getFileName());
+            if(newFile.exists()){
+                fileList.add(streamProcess.getFileName());
+            }
+        }
+
+
+        if(fileList.isEmpty()){
+            return null;
+        }
+        String command = "";
+        outFileName = targetPath + "/" +outFileName;
+        if(fileList.size() == 1){
+             command = String.format(
+                     ffmpeg+ " -i %s -c copy %s",
+                    fileList.get(0), outFileName);
+        }else {
+            String filePaths = String.join("|", fileList);
+            command = String.format(
+                    ffmpeg+ " -i \"concat:%s\" -c copy %s",
+                    filePaths, outFileName);
+        }
+        logger.info("mergeStream {}", command);
+        ProcessBuilder pb = new ProcessBuilder(command.split(" "));
+        pb.redirectErrorStream(true);
+        Process process = pb.start();
+
+        loggingService.execute(new Runnable() {
+            @Override
+            public void run() {
+                try (BufferedReader reader = new BufferedReader(
+                        new InputStreamReader(process.getInputStream()))) {
+                    String line;
+                    while ((line = reader.readLine()) != null) {
+                        logger.info("mergeStream-------- {}",line);
+                    }
+                } catch (IOException e) {
+                    logger.info("mergeStream-------- Over");
+                }
+            }
+        });
+
+
+        return new StreamProcess(process,outFileName,ProcessType.MERGE);
+
+    };
+
+
+}

src/main/java/com/tuoheng/steam/util/FileUtil.java

@@ -0,0 +1,100 @@
+package com.tuoheng.steam.util;
+
+import com.tuoheng.steam.controller.StreamRecordController;
+import com.tuoheng.steam.service.dos.DayRecord;
+import com.tuoheng.steam.service.dos.StreamType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+public class FileUtil {
+
+    private static final Logger logger = LoggerFactory.getLogger(FileUtil.class);
+
+    private  static ScheduledExecutorService executor = Executors.newScheduledThreadPool(4);
+
+    public static void deleteFile(File file) {
+        if(file.exists()) {
+            file.delete();
+        }
+    }
+
+    public static void deleteFolder(String stringPath) {
+
+        try {
+            Path path = Paths.get(stringPath);
+            if (Files.exists(path)) {
+                try (Stream<Path> pathStream = Files.walk(path)) {
+                    pathStream.sorted((p1, p2) -> -p1.compareTo(p2)) // 从子文件/文件夹开始删除
+                            .forEach(p -> {
+                                try {
+                                    Files.delete(p); // 删除文件或文件夹
+                                } catch (IOException e) {
+                                    System.err.println("无法删除: " + p + ", 原因: " + e.getMessage());
+                                }
+                            });
+                }
+            } else {
+                logger.info("文件夹不存在: " + path);
+            }
+        }catch (Exception e) {
+            logger.error("deleteFolder",e);
+        }
+
+    }
+
+    public static List<String> readFiles(String stringPath) {
+
+        List<String> list = new ArrayList<>();
+
+        File directory = new File(stringPath);
+
+        // 检查目录是否存在且是一个目录
+        if (directory.exists() && directory.isDirectory()) {
+            // 获取目录下的所有子文件和文件夹
+            File[] files = directory.listFiles();
+
+            if (files != null) {
+                for (File file : files) {
+                    list.add(file.getName());
+                }
+            }
+        }
+        return list;
+    }
+
+    public static  boolean  isTempFile(String fileName) {
+        String[] result = fileName.split("\\.");
+        return result.length == 3;
+    }
+
+    public static  boolean  isFlv(String fileName) {
+        return fileName.endsWith("flv");
+    }
+
+    public static  boolean  isMp4(String fileName) {
+        return fileName.endsWith("mp4");
+    }
+
+    public static StreamType getStreamType(String fileName) {
+        if(fileName.endsWith("outer")){
+            return StreamType.Outer;
+        }else if(fileName.endsWith("inner")){
+            return StreamType.Inner;
+        }else{
+            return StreamType.LiveVideo;
+        }
+    }
+
+}

src/main/java/com/tuoheng/steam/util/ProcessManager.java

@@ -0,0 +1,110 @@
+package com.tuoheng.steam.util;
+
+import com.tuoheng.steam.controller.StreamRecordController;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+
+import java.io.*;
+import java.util.*;
+
+@Service
+public class ProcessManager {
+
+    private static final Logger logger = LoggerFactory.getLogger(ProcessManager.class);
+
+    private static final String DELIMITER = "=";
+
+    static  Map<Long, String> runningProcessIds = new HashMap<>();
+
+    static String pidPath ="pid.txt";
+
+    static {
+        runningProcessIds = loadFromFile(pidPath);
+        for(Map.Entry<Long, String> entry : runningProcessIds.entrySet()){
+            try {
+                killProcessByPID(entry.getKey());
+            }catch (Exception e){
+                e.printStackTrace();
+            }
+        }
+        runningProcessIds.clear();
+        writeToFile(runningProcessIds);
+    }
+
+    public static  void  registerProcess(Long process) {
+        runningProcessIds.put(process,TimeUtils.formatDateToString(new Date()));
+        writeToFile(runningProcessIds);
+    }
+
+    public static  void unRegisterProcess(Long process) {
+        runningProcessIds.remove(process);
+        writeToFile(runningProcessIds);
+    }
+
+    public static void writeToFile(Map<Long, String> map){
+        try (BufferedWriter writer = new BufferedWriter(new FileWriter(pidPath))) {
+            for (Map.Entry<Long, String> entry : map.entrySet()) {
+                writer.write(entry.getKey() + DELIMITER + entry.getValue());
+                writer.newLine();
+            }
+        }catch (IOException ignored){}
+    }
+
+    public static void appendToFile(Long pid,String strDate)  {
+
+        try (BufferedWriter writer = new BufferedWriter(new FileWriter(pidPath,true))) {
+                writer.write(pid+ DELIMITER + strDate);
+                writer.newLine();
+        }catch (IOException ignored) {}
+    }
+
+    public static Map<Long, String> loadFromFile(String filePath){
+        Map<Long, String> map = new LinkedHashMap<>();
+        File file = new File(filePath);
+        try {
+            if(!file.exists()){
+                file.createNewFile();
+            }
+        }catch (Exception e){
+            e.printStackTrace();
+        }
+        try (BufferedReader reader = new BufferedReader(new FileReader(filePath))) {
+            String line;
+            while ((line = reader.readLine()) != null) {
+                String[] parts = line.split(ProcessManager.DELIMITER, 2); // Split into key and value
+                if (parts.length == 2) {
+                    Long key = Long.parseLong(parts[0].trim());
+                    String value = parts[1].trim();
+                    map.put(key, value);
+                }
+            }
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+        return map;
+    }
+
+    public static void killProcessByPID(long pid) throws IOException, InterruptedException {
+        String os = System.getProperty("os.name").toLowerCase();
+        ProcessBuilder processBuilder;
+
+        if (os.contains("win")) {
+            // Windows: Use taskkill command
+            processBuilder = new ProcessBuilder("cmd", "/c", "taskkill /PID " + pid + " /F");
+        } else {
+            // Linux/Unix/macOS: Use kill command
+            processBuilder = new ProcessBuilder("bash", "-c", "kill -9 " + pid);
+        }
+
+        Process process = processBuilder.start();
+        int exitCode = process.waitFor();
+
+        if (exitCode == 0) {
+            logger.info("Process with PID " + pid + " terminated successfully.");
+        } else {
+            logger.error("Failed to terminate process with PID " + pid + ". Exit code: " + exitCode);
+        }
+    }
+
+}

src/main/java/com/tuoheng/steam/util/TimeUtils.java

@@ -0,0 +1,32 @@
+package com.tuoheng.steam.util;
+
+import java.text.SimpleDateFormat;
+import java.time.LocalDate;
+import java.time.format.DateTimeFormatter;
+import java.util.Date;
+
+public class TimeUtils {
+
+    private static final String DATE_FORMAT = "yyyy-MM-dd HH:mm:ss";
+
+    public static String formatDateToString(Date date) {
+        SimpleDateFormat sdf = new SimpleDateFormat(DATE_FORMAT);
+        return sdf.format(date);
+    }
+
+    private static final DateTimeFormatter DAY_FORMAT = DateTimeFormatter.ofPattern("yyyyMMdd");
+
+    public static boolean isBefore(String dateStr,Integer days) {
+
+        // 将字符串解析为 LocalDate 对象
+        LocalDate date = LocalDate.parse(dateStr, DAY_FORMAT);
+        // 获取当前日期
+        LocalDate currentDate = LocalDate.now();
+        // 判断日期是否在 N 天之前
+        if (date.isBefore(currentDate.minusDays(days))) {
+           return true;
+        } else {
+           return false;
+        }
+    }
+}

src/main/resources/application.properties

@@ -1,16 +1,52 @@
-spring.application.name=demo
+#公司环境
-server.port = 7799
+#spring.application.name=stream_server
-
+#server.port = 9011
-spring.cloud.consul.host=localhost
+#srs.splitPath=/data/java/srs/stream_server/temp
-spring.cloud.consul.port=8500
+#srs.targetPath=/data/java/srs/srs/trunk/objs/nginx/html/recording
-#spring.cloud.consul.discovery.enabled=false 依然会注册上consul，但是不会利用consul调用下游
+#ffmpeg=ffmpeg
-#spring.cloud.consul.discovery.enabled=false
+#recordPath=/data/java/srs/srs/trunk/objs/nginx/html/record
-
+#livedates=8
-spring.cloud.consul.discovery.instance-id= ${spring.application.name}:${vcap.application.instance_id:${spring.application.instance_id:${random.value}}}
+#cangneiwai=false
-
+#srs.domain =  ""
-#下面这部分可以不要
+#srs.name =  ""
-#spring.cloud.consul.discovery.health-check-path=/actuator/health
+#
-#spring.cloud.consul.discovery.health-check-interval=10s
+#大数据局
-#spring.cloud.consul.discovery.health-check-timeout=10s
+#spring.application.name=stream_server
+#server.port = 8989
+#srs.splitPath=/data/java/srs/stream_server/temp
+#srs.targetPath=/data/java/srs/srs/trunk/objs/nginx/html/recording
+#ffmpeg=ffmpeg
+#recordPath=/data/java/srs/srs/trunk/objs/nginx/html/record
+#livedates=8
+#cangneiwai=true
+#srs.domain =  ""
+#srs.name =  ""
 
+#本地测试
+#spring.application.name=stream_server
+#server.port = 9011
+#srs.splitPath=/Users/sunpeng/workspace/stream/temp
+#srs.targetPath=/Users/sunpeng/workspace/stream/html
+#ffmpeg=ffmpeg
+#recordPath=/Users/sunpeng/workspace/stream/record
+#livedates=7
+#cangneiwai=false
+#srs.domain=
+#srs.name=
 
+##容器化部署
+#通过注入
+srs.domain = STREAM.t-aaron.com
+#通过注入
+srs.name = STREAM
+spring.application.name=stream_server
+server.port = 8080
+#零时文件
+srs.splitPath=/data/temp
+#拍照 + 录像
+srs.targetPath=/data/recording
+ffmpeg=ffmpeg
+#
+recordPath=/data/record
+livedates=8
+cangneiwai=true

src/main/resources/logback-spring.xml

@@ -0,0 +1,40 @@
+<configuration>
+	<!-- 控制台输出 -->
+	<appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
+		<encoder>
+			<pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
+		</encoder>
+	</appender>
+
+	<!-- 文件输出（按日滚动） -->
+	<appender name="FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
+		<file>./logs/app.log</file>
+		<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
+			<fileNamePattern>./logs/app.%d{yyyy-MM-dd}.log</fileNamePattern>
+			<maxHistory>7</maxHistory>
+		</rollingPolicy>
+		<encoder>
+			<pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
+		</encoder>
+	</appender>
+	<!--skywalking日志上报-->
+	<appender name="grpc-log" class="org.apache.skywalking.apm.toolkit.log.logback.v1.x.log.GRPCLogClientAppender">
+		<encoder class="ch.qos.logback.core.encoder.LayoutWrappingEncoder">
+			<layout class="org.apache.skywalking.apm.toolkit.log.logback.v1.x.mdc.TraceIdMDCPatternLogbackLayout">
+				<Pattern>[th-airport] %d{yyyy-MM-dd HH:mm:ss.SSS} [%X{tid}] [%thread] %-5level %logger{36} -%msg%n</Pattern>
+			</layout>
+		</encoder>
+	</appender>
+
+	<!-- 日志级别设置 -->
+	<root level="INFO">
+		<appender-ref ref="CONSOLE"/>
+		<appender-ref ref="FILE"/>
+		<appender-ref ref="grpc-log"/>
+	</root>
+<!--	<root level="TRACE">-->
+<!--		<appender-ref ref="CONSOLE"/>-->
+<!--		<appender-ref ref="FILE"/>-->
+<!--		<appender-ref ref="grpc-log"/>-->
+<!--	</root>-->
+</configuration>

src/test/java/com/tuoheng/steam/StreamApplicationTests.java

@@ -1,34 +0,0 @@
-package com.tuoheng.steam;
-
-import com.tuoheng.steam.service.TaskService;
-import org.junit.jupiter.api.Test;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.test.context.SpringBootTest;
-
-@SpringBootTest
-class StreamApplicationTests {
-
-    @Autowired
-    TaskService taskService;
-
-    @Test
-    void contextLoads() {
-        taskService.startTask("rtmp://stream.t-aaron.com/live/123");
-
-        boolean stop = false;
-
-        while (!stop) {
-
-            try {
-                Thread.sleep(60000L);
-            }catch (Exception e){
-
-            }
-
-        }
-
-
-        taskService.stopAllTask();
-    }
-
-}