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:miniotest
Branches Tags
sunpeng:dashujuprod
sunpeng:main
sunpeng:miniotest
sunpeng:consulcheck

3 Commits
main ... miniotest

Author SHA1 Message Date
孙小云 4820bab91d xx 2025-03-15 09:31:34 +08:00
孙小云 70c658ea8c xx 2025-03-13 10:43:03 +08:00
孙小云 785a4cd6d9 xx 2025-03-13 09:14:20 +08:00
15 changed files with 576 additions and 435 deletions
Show Stats Expand all files Collapse all files

23
pom.xml
View File

@ -35,6 +35,11 @@
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.minio</groupId>
@ -46,21 +51,19 @@
<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>io.minio</groupId>-->
<!-- <artifactId>minio</artifactId>-->
<!-- <version>3.0.10</version>-->
<!-- </dependency>-->
<dependency>
<groupId>org.apache.skywalking</groupId>
<artifactId>apm-toolkit-trace</artifactId>
<version>8.6.0</version>
<groupId>io.minio</groupId>
<artifactId>minio</artifactId>
<version>8.3.3</version>
</dependency>
</dependencies>
<build>

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

@ -16,7 +16,6 @@ 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.web.bind.annotation.*;
import java.io.File;
@ -36,86 +35,55 @@ public class StreamRecordController {
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);
System.out.println("启动录制 :"+streamUrl);
if(Objects.isNull(streamUrl)) {
return Response.fail(-1);
}
Response<StreamTask> response = Response.success(taskService.startTask(streamUrl));
logger.info("启动录制返回 :"+ JSON.toJSONString(response));
System.out.println("启动录制返回 :"+ JSON.toJSONString(response));
return response;
}
@GetMapping("stop")
public Response<StreamTask> stopRecording(@RequestParam String streamUrl){
logger.info("关闭录制 :"+streamUrl);
streamUrl = streamSwitch(streamUrl);
logger.info("关闭录制_ :"+streamUrl);
System.out.println("关闭录制 :"+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()){
if(Objects.isNull(response.getData().getOutFileName()) || response.getData().getOutFileName().isEmpty()){
response.setCode(500);
}
logger.info("关闭录制返回 :"+ JSON.toJSONString(response));
System.out.println("关闭录制返回 :"+ 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);
System.out.println("查看录制 :"+streamUrl);
if(Objects.isNull(streamUrl)) {
return Response.fail(-1);
}
Response<StreamTask> response = Response.success(taskService.getLastTask(streamUrl));
logger.info("查看录制返回 :"+ JSON.toJSONString(response));
System.out.println("查看录制返回 :"+ JSON.toJSONString(response));
return response;
}
/**
* 废弃
* @param request
* @return
*/
@PostMapping("search")
public Response<PageInfo<Mp4Info>> streamView(@RequestBody PageStreamRequest request){
logger.info("查看录制 search :"+JSON.toJSONString(request));
System.out.println("查看录制 search :"+JSON.toJSONString(request));
if(Objects.isNull(request.getPageIndex()) || Objects.isNull(request.getPageSize())
|| request.getPageIndex() <0 || request.getPageSize() <=0 ){
logger.error("查看录制 search 入参错误！");
System.out.println("查看录制 search 入参错误！");
return Response.fail(-100);
}
@ -139,10 +107,18 @@ public class StreamRecordController {
*/
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));
System.out.println("查看录制 search 返回:"+JSON.toJSONString(request));
return Response.success(pageInfo);
}
@GetMapping("history")
public Response<List<Mp4Info>> streamView(@RequestParam String streamUrl){
System.out.println("查看录像历史返回 :"+ streamUrl);
List<Mp4Info> dayMp4 = searchAll(streamUrl);
dayMp4.sort((o1, o2) -> o2.getStartTime().compareTo(o1.getStartTime()));
System.out.println("查看录像历史返回 :"+ JSON.toJSONString(dayMp4));
return Response.success(dayMp4) ;
}
public List<Mp4Info> searchAll(String streamUrl){
@ -169,62 +145,22 @@ public class StreamRecordController {
}
@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
*/
@GetMapping("exit")
public String exit(){
Thread thread = new Thread(() -> {
try {
taskService.stopAllTask();
Thread.sleep(10000L);
}catch (Exception e){
}
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;
System.exit(0);
});
return "OK";
}
}

214
src/main/java/com/tuoheng/steam/controller/TestController.java Normal file
View File

@ -0,0 +1,214 @@
package com.tuoheng.steam.controller;
import io.minio.MinioClient;
import io.minio.PutObjectArgs;
import io.minio.credentials.AssumeRoleProvider;
import io.minio.credentials.Credentials;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
@RestController()
@RequestMapping("/test")
public class TestController {
@Value("${minio.oss.endpoint}")
private String endpoint;
@Value("${minio.oss.accessKeyId}")
private String accessKeyId;
@Value("${minio.oss.accessKeySecret}")
private String accessKeySecret;
@Value("${minio.oss.bucketName}")
private String bucketName;
//文件存储目录
@Value("${minio.oss.filedir}")
private String filedir;
@Value("${minio.oss.dajiangName}")
private String dajiangName;
@Value("${minio.oss.dajiangPassword}")
private String dajiangPassword;
//授权策略，允许访问名为bucket的桶的目录
public static final String ROLE_ARN = "arn:aws:s3:::";
private static final Logger logger = LoggerFactory.getLogger(TestController.class);
@GetMapping("/hello")
public String home() {
return "Hello World";
}
@GetMapping("/check")
public String check() throws Exception {
File file = new File("/Users/sunpeng/workspace/text.txt");
InputStream inputStream = new FileInputStream(file);
try {
uploadFile2OSS(inputStream,"text.txt",file.length());
}catch (Exception e) {
logger.error("uploadFile2OSS",e);
}
return "OK";
}
@GetMapping("/key")
public Credentials key() throws Exception {
return getCredentials();
}
private void uploadFile2OSS(InputStream inputStream, String fileName, long streamSize) throws Exception{
MinioClient minioClient = null;
try {
minioClient = MinioClient.builder()
.endpoint(endpoint) // MinIO 服务器地址
.credentials(accessKeyId, accessKeySecret) // 访问密钥和秘密密钥
.build();
}catch (Exception e) {
logger.error("builder",e);
}
String contentType = getContentType(fileName.substring(fileName.lastIndexOf("."))); // 获取文件类型
try {
minioClient.putObject(
PutObjectArgs.builder()
.bucket(bucketName) // 存储桶名称
.object(filedir + "/" + fileName) // 对象名称（路径）
.stream(inputStream, streamSize, -1) // 输入流、文件大小（-1 表示未知大小）
.contentType(contentType) // 文件类型
.build()
);
}catch (Exception e) {
logger.error("putObject",e);
logger.info("bucketName:{}",bucketName);
logger.info("object:{}",filedir + "/" + fileName);
logger.info("inputStream:{}",inputStream);
logger.info("contentType:{}",contentType);
}
}
private static String getContentType(String FilenameExtension) {
if (FilenameExtension.equalsIgnoreCase(".bmp")) {
return "image/bmp";
}
if (FilenameExtension.equalsIgnoreCase(".gif")) {
return "image/gif";
}
if (FilenameExtension.equalsIgnoreCase(".jpeg") ||
FilenameExtension.equalsIgnoreCase(".jpg") ||
FilenameExtension.equalsIgnoreCase(".png")) {
return "image/jpeg";
}
if (FilenameExtension.equalsIgnoreCase(".html")) {
return "text/html";
}
if (FilenameExtension.equalsIgnoreCase(".txt")) {
return "text/plain";
}
if (FilenameExtension.equalsIgnoreCase(".vsd")) {
return "application/vnd.visio";
}
if (FilenameExtension.equalsIgnoreCase(".pptx") ||
FilenameExtension.equalsIgnoreCase(".ppt")) {
return "application/vnd.ms-powerpoint";
}
if (FilenameExtension.equalsIgnoreCase(".docx") ||
FilenameExtension.equalsIgnoreCase(".doc")) {
return "application/msword";
}
if (FilenameExtension.equalsIgnoreCase(".xml")) {
return "text/xml";
}
//PDF
if (FilenameExtension.equalsIgnoreCase(".pdf")) {
return "application/pdf";
}
//excel
if (FilenameExtension.equalsIgnoreCase(".xls") ||
FilenameExtension.equalsIgnoreCase(".xlsx")) {
return "application/octet-stream";
}
//waypoints 拓恒+大疆的航线文件类型
if (FilenameExtension.equalsIgnoreCase(".waypoints") ||
FilenameExtension.equalsIgnoreCase(".kmz")) {
return "application/octet-stream";
}
return "image/jpeg";
}
/**
* 得到 临时凭据
*/
private Credentials getCredentials() throws Exception {
String POLICY_GET_AND_PUT = "{\n" +
" \"Version\": \"2012-10-17\",\n" +
" \"Statement\": [\n" +
" {\n" +
" \"Effect\": \"Allow\",\n" +
" \"Action\": [\n" +
" \"s3:GetObject\",\n" +
" \"s3:GetBucketLocation\",\n" +
" \"s3:PutObject\"\n" +
" ],\n" +
" \"Resource\": [\n" +
" \"arn:aws:s3:::"+bucketName+"/*\"\n" +
" ]\n" +
" }\n" +
" ]\n" + "}";
int durationSeconds = 360000;//秒
//创建签名对象
AssumeRoleProvider provider = new AssumeRoleProvider(
endpoint,
dajiangName,
dajiangPassword,
durationSeconds,//默认3600秒失效，设置小于这个就是3600，大于3600就实际值
POLICY_GET_AND_PUT,
"us-east-1",
ROLE_ARN+bucketName+"/*",
"anysession",
null,
null);
Credentials credentials = provider.fetch();
/**
* 下面的值按照大疆的要求传给大疆
*/
System.out.println("sessionToken=" + credentials.sessionToken());
System.out.println("accessKey=" + credentials.accessKey());
System.out.println("secretKey=" + credentials.secretKey());
System.out.println("isExpired=" + credentials.isExpired());
return credentials;
}
}

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

@ -13,7 +13,7 @@ public class PageInfo <T> implements Serializable {
this.pageIndex = page;
// 计算分页数据
int fromIndex = (page - 1) * pageSize; // 起始索引
int fromIndex = page * pageSize; // 起始索引
int toIndex = Math.min(fromIndex + pageSize, total); // 结束索引
// 检查页码是否超出范围

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

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

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

@ -7,18 +7,55 @@ 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 io.minio.MinioClient;
import io.minio.PutObjectArgs;
import io.minio.credentials.AssumeRoleProvider;
import io.minio.credentials.Credentials;
import io.minio.errors.*;
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.io.*;
import java.security.InvalidKeyException;
import java.security.NoSuchAlgorithmException;
import java.util.*;
@Component
public class Scheduler {
@Value("${minio.oss.endpoint}")
private String endpoint;
@Value("${minio.oss.accessKeyId}")
private String accessKeyId;
@Value("${minio.oss.accessKeySecret}")
private String accessKeySecret;
@Value("${minio.oss.bucketName}")
private String bucketName;
//文件存储目录
@Value("${minio.oss.filedir}")
private String filedir;
@Value("${minio.oss.dajiangName}")
private String dajiangName;
@Value("${minio.oss.dajiangPassword}")
private String dajiangPassword;
//授权策略，允许访问名为bucket的桶的目录
public static final String ROLE_ARN = "arn:aws:s3:::";
private static final Logger logger = LoggerFactory.getLogger(Scheduler.class);
@Autowired
@ -27,42 +64,158 @@ public class Scheduler {
@Value("${livedates}")
private Integer livedates;
@Value("${cangneiwai}")
private Boolean cangneiwai;
/**
* 初次执行延迟6秒执行
* 每隔 60 分钟执行一次 60*60*1000
*/
@Scheduled(fixedRate = 3600000, initialDelay = 6000)
public void mergeTask() {
@Scheduled(fixedRate = 3600000, initialDelay = 10)
public void mergeTask() throws Exception {
// File file = new File("/Users/sunpeng/workspace/text.txt");
// InputStream inputStream = new FileInputStream(file);
// try {
// uploadFile2OSS(inputStream,"text.txt",file.length());
// }catch (Exception e) {
// logger.error(e.getMessage());
// }
if(!cangneiwai){
logger.info("舱内外无需录制");
return;
getCredentials();
}
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);
}
}
}
}
/**
* 得到 临时凭据
*/
private Credentials getCredentials() throws Exception {
String POLICY_GET_AND_PUT = "{\n" +
" \"Version\": \"2012-10-17\",\n" +
" \"Statement\": [\n" +
" {\n" +
" \"Effect\": \"Allow\",\n" +
" \"Action\": [\n" +
" \"s3:GetObject\",\n" +
" \"s3:GetBucketLocation\",\n" +
" \"s3:PutObject\"\n" +
" ],\n" +
" \"Resource\": [\n" +
" \"arn:aws:s3:::"+bucketName+"/*\"\n" +
" ]\n" +
" }\n" +
" ]\n" + "}";
int durationSeconds = 360000;//秒
//创建签名对象
AssumeRoleProvider provider = new AssumeRoleProvider(
endpoint,
dajiangName,
dajiangPassword,
durationSeconds,//默认3600秒失效，设置小于这个就是3600，大于3600就实际值
POLICY_GET_AND_PUT,
"us-east-1",
ROLE_ARN+bucketName+"/*",
"anysession",
null,
null);
Credentials credentials = provider.fetch();
/**
* 下面的值按照大疆的要求传给大疆
*/
// System.out.println("sessionToken=" + credentials.sessionToken());
// System.out.println("accessKey=" + credentials.accessKey());
// System.out.println("secretKey=" + credentials.secretKey());
// System.out.println("isExpired=" + credentials.isExpired());
return credentials;
}
private void uploadFile2OSS(InputStream inputStream, String fileName, long streamSize) throws Exception{
// 初始化 MinioClient
MinioClient minioClient = MinioClient.builder()
.endpoint(endpoint) // MinIO 服务器地址
.credentials(accessKeyId, accessKeySecret) // 访问密钥和秘密密钥
.build();
String contentType = getContentType(fileName.substring(fileName.lastIndexOf("."))); // 获取文件类型
try {
minioClient.putObject(
PutObjectArgs.builder()
.bucket(bucketName) // 存储桶名称
.object(filedir + "/" + fileName) // 对象名称（路径）
.stream(inputStream, streamSize, -1) // 输入流、文件大小（-1 表示未知大小）
.contentType(contentType) // 文件类型
.build()
);
}catch (Exception e) {
logger.error(e.getMessage());
}
}
private static String getContentType(String FilenameExtension) {
if (FilenameExtension.equalsIgnoreCase(".bmp")) {
return "image/bmp";
}
if (FilenameExtension.equalsIgnoreCase(".gif")) {
return "image/gif";
}
if (FilenameExtension.equalsIgnoreCase(".jpeg") ||
FilenameExtension.equalsIgnoreCase(".jpg") ||
FilenameExtension.equalsIgnoreCase(".png")) {
return "image/jpeg";
}
if (FilenameExtension.equalsIgnoreCase(".html")) {
return "text/html";
}
if (FilenameExtension.equalsIgnoreCase(".txt")) {
return "text/plain";
}
if (FilenameExtension.equalsIgnoreCase(".vsd")) {
return "application/vnd.visio";
}
if (FilenameExtension.equalsIgnoreCase(".pptx") ||
FilenameExtension.equalsIgnoreCase(".ppt")) {
return "application/vnd.ms-powerpoint";
}
if (FilenameExtension.equalsIgnoreCase(".docx") ||
FilenameExtension.equalsIgnoreCase(".doc")) {
return "application/msword";
}
if (FilenameExtension.equalsIgnoreCase(".xml")) {
return "text/xml";
}
//PDF
if (FilenameExtension.equalsIgnoreCase(".pdf")) {
return "application/pdf";
}
//excel
if (FilenameExtension.equalsIgnoreCase(".xls") ||
FilenameExtension.equalsIgnoreCase(".xlsx")) {
return "application/octet-stream";
}
//waypoints 拓恒+大疆的航线文件类型
if (FilenameExtension.equalsIgnoreCase(".waypoints") ||
FilenameExtension.equalsIgnoreCase(".kmz")) {
return "application/octet-stream";
}
return "image/jpeg";
}
// public static boolean isWithin15Minutes(String timestamp1, String timestamp2) {
// // 将字符串转换为 long 类型
// long time1 = Long.parseLong(timestamp1);

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

@ -5,15 +5,14 @@ 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);
}

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

@ -2,16 +2,11 @@ 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 stopTask(String streamUrl);
public StreamTask getLastTask(String streamUrl);

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

@ -4,7 +4,6 @@ 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;
@ -31,8 +30,8 @@ public class TaskService implements ITaskService{
@Value("${srs.targetPath}")
private String targetPath;
// @Value("${ffmpeg}")
private String ffmpeg ="ffmpeg";
@Value("${ffmpeg}")
private String ffmpeg;
ScheduledExecutorService executor = Executors.newScheduledThreadPool(4);
@ -43,20 +42,18 @@ public class TaskService implements ITaskService{
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 ) {
if(streamTask.getStartTime().getTime() < new Date().getTime() - 30 * 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());
file.delete();
logger.info("废弃文件删除成功");
}catch (Exception e) {
logger.error(e.getMessage());
}
}, 60, TimeUnit.SECONDS);
}catch (Exception ignore) {}
}
}
}
@ -96,9 +93,7 @@ public class TaskService implements ITaskService{
Date twoHoursAgo = calendar.getTime();
if (value.getStartTime().before(twoHoursAgo)) {
logger.error("taskId {} 执行超时,手动关闭", key);
try {
stopTask(key);
}catch (Exception ignore) {}
}
});
@ -109,8 +104,9 @@ public class TaskService implements ITaskService{
for (StreamProcess process : streamProcesses) {
process.destroy();
File file = new File(process.getFileName());
FileUtil.deleteFile(file);
if(file.exists()){
file.delete();
}
}
}
@ -122,34 +118,6 @@ public class TaskService implements ITaskService{
}, 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) {
/*
@ -174,9 +142,7 @@ public class TaskService implements ITaskService{
}
public StreamTask stopTask(String streamUrl) throws Exception {
boolean recordSuccess = true;
public StreamTask stopTask(String streamUrl) {
StreamTask currentStreamTask = runningTasks.remove(streamUrl);
if (currentStreamTask!= null) {
@ -184,20 +150,11 @@ public class TaskService implements ITaskService{
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());
@ -220,7 +177,9 @@ public class TaskService implements ITaskService{
ProcessManager.unRegisterProcess(mergeProcess.getInnerProcessId());
for(StreamProcess streamProcess :currentStreamTask.getStreamProcesses()){
File file = new File(streamProcess.getFileName());
FileUtil.deleteFile(file);
if(file.exists()){
file.delete();
}
}
}).exceptionally(ex -> {
logger.error("streamUrl {} taskId {} MergeProcess {} exceptionally", streamUrl,currentStreamTask.getTaskId(), mergeProcess.getInnerProcessId(), ex);
@ -228,7 +187,9 @@ public class TaskService implements ITaskService{
ProcessManager.unRegisterProcess(mergeProcess.getInnerProcessId());
for(StreamProcess streamProcess :currentStreamTask.getStreamProcesses()){
File file = new File(streamProcess.getFileName());
FileUtil.deleteFile(file);
if(file.exists()){
file.delete();
}
}
return null;
});
@ -236,26 +197,35 @@ public class TaskService implements ITaskService{
}
// if( historyTasks.containsKey(streamUrl)){
// historyTasks.get(streamUrl).add(currentStreamTask);
// }else {
// historyTasks.put(streamUrl, new LinkedList<>());
// historyTasks.get(streamUrl).offerLast(currentStreamTask);
// }
if(Objects.nonNull(currentStreamTask.getMergeProcess())){
if(Objects.nonNull(currentStreamTask.getMergeProcess().getFileName()) &&
!currentStreamTask.getMergeProcess().getFileName().isEmpty()){
try {
//
// scheduler.schedule(task, 3, TimeUnit.SECONDS);
executor.schedule(new Runnable() {
@Override
public void run() {
try {
logger.info("FileName :"+currentStreamTask.getMergeProcess().getFileName());
System.out.println("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);
System.out.println("Duration line: " + line);
if (line.contains("Duration")) {
String duration = line.split("Duration: ")[1].split(",")[0];
logger.info("Duration: " + duration.substring(0, 8));
System.out.println("Duration: " + duration.substring(0, 8));
currentStreamTask.setDuration(duration.substring(0, 8));
logger.info("-----------------放入缓存-----------------");
System.out.println("-----------------放入缓存-----------------");
/**
* 放入缓存
*/
@ -270,7 +240,7 @@ public class TaskService implements ITaskService{
}
reader.close();
}catch (Exception e){
logger.info(e.getMessage());
System.out.println(e.getMessage());
}
}
},40, TimeUnit.SECONDS);
@ -289,12 +259,11 @@ public class TaskService implements ITaskService{
}
}else {
throw new Exception("");
// if(historyTasks.containsKey(streamUrl)){
// return historyTasks.get(streamUrl).peekLast();
// }else {
// return null;
// }
if(historyTasks.containsKey(streamUrl)){
return historyTasks.get(streamUrl).peekLast();
}else {
return null;
}
}
}
@ -334,14 +303,8 @@ public class TaskService implements ITaskService{
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);
ProcessManager.unRegisterProcess(streamProcess.getInnerProcessId());
}).exceptionally(ex -> {
logger.info("streamUrl {} taskId {} Process {} Exceptionally", streamUrl, streamTask.getTaskId(),streamProcess.getInnerProcessId(),ex);
streamProcess.setStopTime(TimeUtils.formatDateToString(new Date()));
@ -359,9 +322,7 @@ public class TaskService implements ITaskService{
public void stopAllTask(){
runningTasks.forEach((key, value) -> {
try {
stopTask(key);
}catch (Exception ignore){}
});
}
}

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

@ -3,7 +3,6 @@ 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;
@ -15,8 +14,14 @@ 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.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;
@Service
@ -25,8 +30,8 @@ public class ProcessService {
@Value("${srs.splitPath}")
private String splitPath;
// @Value("${ffmpeg}")
private String ffmpeg = "ffmpeg";
@Value("${ffmpeg}")
private String ffmpeg;
@Value("${srs.targetPath}")
private String targetPath;
@ -57,7 +62,6 @@ public class ProcessService {
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;
@ -71,105 +75,34 @@ public class ProcessService {
File delete = new File(recordPath+ File.separator +
flvRecord.getStream().getDayRecord().getDay() + File.separator +
flvRecord.getStream().getStreamId() + File.separator + flvRecord.getFlv());
FileUtil.deleteFile(delete);
if(delete.exists()){
delete.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();
}
}
}
}
Process completedProcess = future.get();
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);
streamUrl, recordFileName);
logger.info("recordStream {}", command);
ProcessBuilder pb = new ProcessBuilder(command.split(" "));
pb.redirectErrorStream(true);

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

@ -1,10 +1,7 @@
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;
@ -13,23 +10,10 @@ 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 {
@ -46,10 +30,10 @@ public class FileUtil {
});
}
} else {
logger.info("文件夹不存在: " + path);
System.out.println("文件夹不存在: " + path);
}
}catch (Exception e){
logger.error("deleteFolder",e);
e.printStackTrace();
}
}
@ -69,7 +53,11 @@ public class FileUtil {
for (File file : files) {
list.add(file.getName());
}
} else {
}
} else {
}
return list;
}

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

@ -1,8 +1,5 @@
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.*;
@ -11,8 +8,6 @@ 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<>();
@ -101,9 +96,9 @@ public class ProcessManager {
int exitCode = process.waitFor();
if (exitCode == 0) {
logger.info("Process with PID " + pid + " terminated successfully.");
System.out.println("Process with PID " + pid + " terminated successfully.");
} else {
logger.error("Failed to terminate process with PID " + pid + ". Exit code: " + exitCode);
System.err.println("Failed to terminate process with PID " + pid + ". Exit code: " + exitCode);
}
}

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

@ -1,52 +1,22 @@
#公司环境
#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 = 7788
srs.splitPath=/data/java/srs/stream_server/temp
srs.targetPath=/data/java/srs/srs/trunk/objs/nginx/html
ffmpeg=/data/ffmpeg/bin/ffmpeg
recordPath=/data/java/srs/srs/trunk/objs/nginx/html/record
livedates=8
#本地测试
#spring.application.name=stream_server
#server.port = 9011
#server.port = 8080
#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
minio.oss.endpoint: https://minio-jndsj.t-aaron.com:2443
minio.oss.accessKeyId: PJM0c2qlauoXv5TMEHm2
minio.oss.accessKeySecret: Wr69Dm3ZH39M3GCSeyB3eFLynLPuGCKYfphixZuI
minio.oss.bucketName: th-airport
minio.oss.filedir: prodFile
minio.oss.dajiangName: dajiang
minio.oss.dajiangPassword: dajiang2025

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();
}
}