sunpeng
/
stream_srs
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Compare commits

base: sunpeng:main
Branches Tags
sunpeng:main
sunpeng:dashujuprod
sunpeng:miniotest
sunpeng:consulcheck
..
compare: sunpeng:consulcheck
Branches Tags
sunpeng:dashujuprod
sunpeng:main
sunpeng:miniotest
sunpeng:consulcheck

4 Commits
main ... consulchec

Author SHA1 Message Date
孙小云 77c31ce49b xx 2025-03-13 16:12:45 +08:00
孙小云 4f6df96530 xx 2025-03-13 10:36:20 +08:00
孙小云 5a4d0e2404 xx 2025-03-13 09:49:12 +08:00
孙小云 785a4cd6d9 xx 2025-03-13 09:14:20 +08:00
26 changed files with 84 additions and 2028 deletions
Show Stats Expand all files Collapse all files

59
pom.xml
View File

@ -5,7 +5,7 @@
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.7.18</version>
<version>2.3.5.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.example</groupId>
@ -13,56 +13,46 @@
<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>io.minio</groupId>
<artifactId>minio</artifactId>
<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>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<!--健康检查端点-->
<dependency>
<groupId>org.apache.skywalking</groupId>
<artifactId>apm-toolkit-trace</artifactId>
<version>8.6.0</version>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</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>
@ -72,4 +62,5 @@
</plugins>
</build>
</project>

227
src/main/java/com/tuoheng/steam/controller/StreamRecordController.java
View File

@ -1,230 +1,25 @@
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.beans.factory.annotation.Value;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.discovery.DiscoveryClient;
import org.springframework.web.bind.annotation.*;
import java.io.File;
import java.util.*;
import java.util.stream.Collectors;
import java.util.List;
@RestController()
@RequestMapping("/record")
public class StreamRecordController {
private static final Logger logger = LoggerFactory.getLogger(StreamRecordController.class);
@Autowired
ITaskService taskService;
private DiscoveryClient discoveryClient;
@Autowired
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("/")
public String home() {
List<ServiceInstance> instances = discoveryClient.getInstances("demo");
List<String> services = discoveryClient.getServices();
return "Hello World!";
}
@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;
}
}

71
src/main/java/com/tuoheng/steam/controller/dto/PageInfo.java
View File

@ -1,71 +0,0 @@
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;
}
}

53
src/main/java/com/tuoheng/steam/controller/dto/PageStreamRequest.java
View File

@ -1,53 +0,0 @@
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;
}
}

37
src/main/java/com/tuoheng/steam/controller/dto/Response.java
View File

@ -1,37 +0,0 @@
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;
}
}

28
src/main/java/com/tuoheng/steam/dos/Mp4Info.java
View File

@ -1,28 +0,0 @@
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;
}
}

5
src/main/java/com/tuoheng/steam/dos/ProcessType.java
View File

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

78
src/main/java/com/tuoheng/steam/dos/StreamProcess.java
View File

@ -1,78 +0,0 @@
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;
}
}

136
src/main/java/com/tuoheng/steam/dos/StreamTask.java
View File

@ -1,136 +0,0 @@
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;
}

78
src/main/java/com/tuoheng/steam/schedule/Scheduler.java
View File

@ -1,78 +0,0 @@
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;
// }
}

19
src/main/java/com/tuoheng/steam/service/IRecordService.java
View File

@ -1,19 +0,0 @@
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);
}

19
src/main/java/com/tuoheng/steam/service/ITaskService.java
View File

@ -1,19 +0,0 @@
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();
}

67
src/main/java/com/tuoheng/steam/service/RecordService.java
View File

@ -1,67 +0,0 @@
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());
}
}
}

367
src/main/java/com/tuoheng/steam/service/TaskService.java
View File

@ -1,367 +0,0 @@
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){}
});
}
}

50
src/main/java/com/tuoheng/steam/service/dos/DayRecord.java
View File

@ -1,50 +0,0 @@
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;
}
}

44
src/main/java/com/tuoheng/steam/service/dos/FlvRecord.java
View File

@ -1,44 +0,0 @@
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);
}
}

45
src/main/java/com/tuoheng/steam/service/dos/Mp4Record.java
View File

@ -1,45 +0,0 @@
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);
}
}

95
src/main/java/com/tuoheng/steam/service/dos/StreamRecord.java
View File

@ -1,95 +0,0 @@
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;
}
}

5
src/main/java/com/tuoheng/steam/service/dos/StreamType.java
View File

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

249
src/main/java/com/tuoheng/steam/service/innerService/ProcessService.java
View File

@ -1,249 +0,0 @@
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);
};
}

100
src/main/java/com/tuoheng/steam/util/FileUtil.java
View File

@ -1,100 +0,0 @@
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;
}
}
}

110
src/main/java/com/tuoheng/steam/util/ProcessManager.java
View File

@ -1,110 +0,0 @@
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);
}
}
}

32
src/main/java/com/tuoheng/steam/util/TimeUtils.java
View File

@ -1,32 +0,0 @@
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;
}
}
}

64
src/main/resources/application.properties
View File

@ -1,52 +1,16 @@
#公司环境
#spring.application.name=stream_server
#server.port = 9011
#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=false
#srs.domain = ""
#srs.name = ""
#
#大数据局
#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=demo
server.port = 7799
spring.cloud.consul.host=localhost
spring.cloud.consul.port=8500
#spring.cloud.consul.discovery.enabled=false 依然会注册上consul，但是不会利用consul调用下游
#spring.cloud.consul.discovery.enabled=false
spring.cloud.consul.discovery.instance-id= ${spring.application.name}:${vcap.application.instance_id:${spring.application.instance_id:${random.value}}}
#下面这部分可以不要
#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 = 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

40
src/main/resources/logback-spring.xml
View File

@ -1,40 +0,0 @@
<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>

34
src/test/java/com/tuoheng/steam/StreamApplicationTests.java Normal file
View File

@ -0,0 +1,34 @@
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();
}
}