前言
前面的文章中使用websocket的方案在web端实现rtsp播放，因为各种原因，现需要重新写一套方案。不废话，上才艺！！！
补充：
-
项目中需求可能要同时观看多个摄像头；将本项目放开限制使用多个摄像头时，就会发现相机之间的切换加载时间及视频流畅度明显降（个人判断主要是网络及带宽的影响导致）
-
本人尝试的第三种方式用docker镜像通过rancher-api启动单个实例对应单个相机功能也已经实现；预计后期项目中会采用第三套方案；该方案暂不打算全部公开，后续会写相关博客介绍思路，感兴趣的可关注，需要的可以联系博主QQ：1007026004，有偿，介意勿扰！
添加依赖
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>io.springfox</groupId>
<artifactId>springfox-swagger-ui</artifactId>
<version>2.4.0</version>
</dependency>
<dependency>
<groupId>io.springfox</groupId>
<artifactId>springfox-swagger2</artifactId>
<version>2.4.0</version>
</dependency>
<dependency>
<groupId>io.minio</groupId>
<artifactId>minio</artifactId>
<version>8.1.0</version>
</dependency>
<dependency>
<groupId>org.mybatis.spring.boot</groupId>
<artifactId>mybatis-spring-boot-starter</artifactId>
<version>1.3.2</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<!-- https://mvnrepository.com/artifact/org.jetbrains.kotlin/kotlin-stdlib -->
<dependency>
<groupId>org.jetbrains.kotlin</groupId>
<artifactId>kotlin-stdlib</artifactId>
<version>1.3.70</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-thymeleaf</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-configuration-processor</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-websocket</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
<version>${commons.io.version}</version>
</dependency>
<!--文件上传工具类 -->
<dependency>
<groupId>commons-fileupload</groupId>
<artifactId>commons-fileupload</artifactId>
<version>${commons.fileupload.version}</version>
</dependency>
<!-- 阿里JSON解析器 -->
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>${fastjson.version}</version>
</dependency>
<!--好用的工具集-->
<dependency>
<groupId>cn.hutool</groupId>
<artifactId>hutool-all</artifactId>
<version>${hutool.version}</version>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
<version>${lang3.version}</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
<version>${jackson.version}</version>
</dependency>
<dependency>
<groupId>org.bytedeco</groupId>
<artifactId>javacv-platform</artifactId>
<version>1.5.5</version>
</dependency>
<dependency>
<groupId>org.bytedeco</groupId>
<artifactId>opencv-platform</artifactId>
<version>4.5.1-1.5.5</version>
</dependency>
<!-- 支持 @ConfigurationProperties 注解 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-configuration-processor</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>javax.validation</groupId>
<artifactId>validation-api</artifactId>
</dependency>
<dependency>
<groupId>org.hibernate.validator</groupId>
<artifactId>hibernate-validator</artifactId>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-exec</artifactId>
<version>1.3</version>
</dependency>
创建实体
@Data
@Accessors(chain = true)
@ToString
@NoArgsConstructor
@AllArgsConstructor
public class CameraBean implements Serializable {
private static final long serialVersionUID = 8183688502930584159L;
@NotEmpty(message = "不能为空")
@ApiModelProperty(value = "摄像头ip",required = true)
private String cameraIp;
@NotEmpty(message = "不能为空")
@ApiModelProperty(value = "账号",required = true)
private String cameraAccount;
@NotEmpty(message = "不能为空")
@ApiModelProperty(value = "密码",required = true)
private String cameraPassword;
@ApiModelProperty(value = "rtsp地址",hidden = true)
private String rtsp;
@ApiModelProperty(value = "rtmp地址",hidden = true)
private String rtmp;
@ApiModelProperty(value = "打开时间",hidden = true)
private String openTime;
@ApiModelProperty(value = "使用人数",hidden = true)
private int count = 0;
@ApiModelProperty(value = "token",hidden = true)
private String token;
}
控制层
开启直播与关闭直播
@Slf4j
@RestController
public class CameraVideoController {
/*
* 保存已经开始推的流 streamMap
*/
public static Map<String, CameraBean> STREAMMAP = new ConcurrentHashMap<>();
@Value("${push.host:127.0.0.1}")
private String pushHost;
@Value("${push.port:1935}")
private String pushPort;
// 存放任务 线程
public static Map<String, CameraThread.MyRunnable> JOBMAP = new HashMap<>();
@ApiOperation(value = "开启视频流")
@RequestMapping(value = "/openVideo", method = RequestMethod.POST)
public R openVideo(@RequestBody @Valid CameraBean vo) {
if (!Utils.isTrueIp(vo.getCameraIp())) {
return R.fail("ip格式输入错误");
}
// 获取当前时间
String openTime = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date().getTime());
Set<String> keys = STREAMMAP.keySet();
// 当推流map为空时才让进入转封装，避免资源不够
if (0 == keys.size()) {
// 开始推流
String token = UUID.randomUUID().toString();
String IP = Utils.IpConvert(vo.getCameraIp());
String url = "";
// 直播流
String rtsp = "rtsp://" + vo.getCameraAccount() + ":" + vo.getCameraPassword() + "@" + IP + ":554/cam/realmonitor?channel=1&subtype=0";
String rtmp = "rtmp://" + Utils.IpConvert(pushHost) + ":" + pushPort + "/live/" + token;
vo.setRtsp(rtsp);
vo.setRtmp(rtmp);
vo.setOpenTime(openTime);
vo.setToken(token);
vo.setCount(1);
// 先验证ip是否能连接成功，避免后续推流失败
Socket rtspSocket = new Socket();
Socket rtmpSocket = new Socket();
// 建立TCP Scoket连接，超时时间1s *3
try {
rtspSocket.connect(new InetSocketAddress(vo.getCameraIp(), 554), 1000*3);
} catch (IOException e) {
e.printStackTrace();
log.error("与拉流IP：" + vo.getCameraIp() + " 建立TCP连接失败！");
return R.data(500,vo,"与拉流IP：" + vo.getCameraIp() +" 建立TCP连接失败！");
}
try {
rtmpSocket.connect(new InetSocketAddress(Utils.IpConvert(pushHost),Integer.parseInt(pushPort)), 1000*3);
} catch (IOException e) {
e.printStackTrace();
log.error("与推流IP： " + pushHost + " 端口： " + pushPort + " 建立TCP连接失败！");
return R.data(500,vo,"与推流IP：" + pushHost +" 建立TCP连接失败,请检查nginx服务");
}
// 执行任务
CameraThread.MyRunnable task = new CameraThread.MyRunnable(vo);
CameraThread.MyRunnable.threadPool.execute(task);
JOBMAP.put(token, task);
}
return R.data(200,vo.getRtmp(),"成功！");
}
@ApiOperation(value = "关闭视频流")
@RequestMapping(value = "/closeVideo/{token}", method = RequestMethod.GET)
public R closeVideo(@PathVariable("token") String token) {
if (StringUtils.isNotEmpty(token)) {
if (JOBMAP.containsKey(token) && STREAMMAP.containsKey(token)) {
if (0 < STREAMMAP.get(token).getCount()) {
// 人数-1
STREAMMAP.get(token).setCount(STREAMMAP.get(token).getCount() - 1);
CameraVideoController.JOBMAP.get(token).setInterrupted(token);
log.info("关闭成功 当前相机使用人数为" + STREAMMAP.get(token).getCount() + " [ip："+ STREAMMAP.get(token).getCameraIp() +" rtmp:" + STREAMMAP.get(token).getRtmp() + "]");
return R.success("关闭成功");
}
}
}
return R.fail("请检查该token下的视频是否开启成功");
}
}
推流
rtsp转封装到rtmp关键类
@Slf4j
@Data
public class RtmpPush {
/*
* 保存push
*/
public static Map<String, RtmpPush> RTMPPUSHMAP = new ConcurrentHashMap<>();
private CameraBean cameraBean;
private FFmpegFrameRecorder recorder;
private FFmpegFrameGrabber grabber;
// 推流过程中出现错误的次数
private int errIndex = 0;
// 退出状态码：0-正常退出;1-手动中断;
private int exitCode = 0;
// 帧率
private double frameRate = 0;
public RtmpPush(CameraBean cameraBean) {
this.cameraBean = cameraBean;
}
/**
* 推送视频流数据包
**/
public void push() {
try {
avutil.av_log_set_level(avutil.AV_LOG_INFO);
FFmpegLogCallback.set();
grabber = new FFmpegFrameGrabber(cameraBean.getRtsp());
grabber.setOption("rtsp_transport", "tcp");
// 设置采集器构造超时时间
grabber.setOption("stimeout", "2000000");
grabber.start();
// 部分监控设备流信息里携带的帧率为9000，如出现此问题，会导致dts、pts时间戳计算失败，播放器无法播放，故出现错误的帧率时，默认为25帧
if (grabber.getFrameRate() > 0 && grabber.getFrameRate() < 100) {
frameRate = grabber.getFrameRate();
} else {
frameRate = 25.0;
}
int width = grabber.getImageWidth();
int height = grabber.getImageHeight();
// 若视频像素值为0，说明拉流异常，程序结束
if (width == 0 && height == 0) {
log.error(cameraBean.getRtsp() + " 拉流异常！");
grabber.stop();
grabber.close();
release();
return;
}
recorder = new FFmpegFrameRecorder(cameraBean.getRtmp(), grabber.getImageWidth(), grabber.getImageHeight());
recorder.setInterleaved(true);
// 关键帧间隔，一般与帧率相同或者是视频帧率的两倍
recorder.setGopSize((int) frameRate * 2);
// 视频帧率(保证视频质量的情况下最低25，低于25会出现闪屏)
recorder.setFrameRate(frameRate);
// 设置比特率
recorder.setVideoBitrate(grabber.getVideoBitrate());
// 封装flv格式
recorder.setFormat("flv");
// h264编/解码器
recorder.setVideoCodec(avcodec.AV_CODEC_ID_H264);
recorder.setPixelFormat(avutil.AV_PIX_FMT_YUV420P);
Map<String, String> videoOption = new HashMap<>();
// 该参数用于降低延迟
videoOption.put("tune", "zerolatency");
/**
** 权衡quality(视频质量)和encode speed(编码速度) values(值)： *
* ultrafast(终极快),superfast(超级快), veryfast(非常快), faster(很快), fast(快), *
* medium(中等), slow(慢), slower(很慢), veryslow(非常慢) *
* ultrafast(终极快)提供最少的压缩（低编码器CPU）和最大的视频流大小；而veryslow(非常慢)提供最佳的压缩（高编码器CPU）的同时降低视频流的大小
*/
videoOption.put("preset", "ultrafast");
// 画面质量参数，0~51；18~28是一个合理范围
videoOption.put("crf", "28");
recorder.setOptions(videoOption);
AVFormatContext fc = grabber.getFormatContext();
recorder.start(fc);
log.debug("开始推流 [rtsp:" + cameraBean.getRtsp() + " rtmp:" + cameraBean.getRtmp() + "]");
// 清空探测时留下的缓存
grabber.flush();
AVPacket pkt = null;
long dts = 0;
long pts = 0;
int timebase = 0;
for (int no_frame_index = 0; no_frame_index < 10 && errIndex < 10;) {
long time1 = System.currentTimeMillis();
if (exitCode == 1) {
break;
}
pkt = grabber.grabPacket();
if (pkt == null || pkt.size() == 0 || pkt.data() == null) {
// 空包记录次数跳过
log.warn("JavaCV 出现空包 [rtsp:" + cameraBean.getRtsp() + " rtmp:" + cameraBean.getRtmp() + "]");
no_frame_index++;
continue;
}
// 过滤音频
if (pkt.stream_index() == 1) {
av_packet_unref(pkt);
continue;
}
// 矫正sdk回调数据的dts，pts每次不从0开始累加所导致的播放器无法续播问题
pkt.pts(pts);
pkt.dts(dts);
errIndex += (recorder.recordPacket(pkt) ? 0 : 1);
// pts,dts累加
timebase = grabber.getFormatContext().streams(pkt.stream_index()).time_base().den();
pts += timebase / (int) frameRate;
dts += timebase / (int) frameRate;
// 将缓存空间的引用计数-1，并将Packet中的其他字段设为初始值。如果引用计数为0，自动的释放缓存空间。
av_packet_unref(pkt);
long endtime = System.currentTimeMillis();
// long num = (long) (1000 / frameRate) - (endtime - time1);
// if (num > 0) {
// Thread.sleep(num);
// }
}
} catch (Exception e) {
e.printStackTrace();
log.error(e.getMessage());
} finally {
release();
log.info("推流结束 [rtsp:" + cameraBean.getRtsp() + " rtmp:" + cameraBean.getRtmp() + "]");
}
}
/**
*资源释放
**/
public void release() {
try {
grabber.stop();
grabber.close();
if (recorder != null) {
recorder.stop();
recorder.release();
}
} catch (Exception e) {
e.printStackTrace();
}
}
}
线程池
线程的管理，开启与中断
@Slf4j
public class CameraThread {
public static class MyRunnable implements Runnable {
private static int core = Runtime.getRuntime().availableProcessors();
// 创建线程池
public static ExecutorService threadPool = new ThreadPoolExecutor(
core, core*4, 3,
TimeUnit.SECONDS,
new LinkedBlockingDeque<>(core),Executors.defaultThreadFactory(),
new ThreadPoolExecutor.DiscardOldestPolicy());
private CameraBean cameraBean;
public MyRunnable(CameraBean cameraBean) {
this.cameraBean = cameraBean;
}
// 中断线程
public void setInterrupted(String key) {
RtmpPush.RTMPPUSHMAP.get(key).setExitCode(1);
}
@Override
public void run() {
// 直播流
try {
CameraVideoController.STREAMMAP.put(cameraBean.getToken(), cameraBean);
// 执行转流推流任务
RtmpPush push = new RtmpPush(cameraBean);
RtmpPush.RTMPPUSHMAP.put(cameraBean.getToken(), push);
push.push();
// 清除缓存
CameraVideoController.STREAMMAP.remove(cameraBean.getToken());
CameraVideoController.JOBMAP.remove(cameraBean.getToken());
RtmpPush.RTMPPUSHMAP.remove(cameraBean.getToken());
} catch (Exception e) {
e.printStackTrace();
CameraVideoController.STREAMMAP.remove(cameraBean.getToken());
CameraVideoController.JOBMAP.remove(cameraBean.getToken());
RtmpPush.RTMPPUSHMAP.remove(cameraBean.getToken());
}
}
}
}
工具类
参数校验工具
@Slf4j
public class Utils {
/**
* 域名转ip
**/
public static String IpConvert(String domainName) {
String ip = domainName;
try {
ip = InetAddress.getByName(domainName).getHostAddress();
} catch (UnknownHostException e) {
e.printStackTrace();
return domainName;
}
return ip;
}
/**
*参数非空校验
**/
public static boolean isNullParameters(JSONObject cameraJson, String[] isNullArr) {
Map<String, Object> checkMap = new HashMap<>();
// 空值校验
for (String key : isNullArr) {
if (null == cameraJson.get(key) || "".equals(cameraJson.get(key))) {
return false;
}
}
return true;
}
/**
*参数ip格式校验
**/
public static boolean isTrueIp(String ip) {
return ip.matches("([1-9]|[1-9]\\d|1\\d{2}|2[0-4]\\d|25[0-5])(\\.(\\d|[1-9]\\d|1\\d{2}|2[0-4]\\d|25[0-5])){3}");
}
}
api工具类
返回相关的类
public interface IResultCode extends Serializable {
String getMessage();
int getCode();
}
@ApiModel(description = "返回信息")
public class R<T> implements Serializable {
private static final long serialVersionUID = 1L;
@ApiModelProperty(value = "状态码",required = true)
private int code;
@ApiModelProperty(value = "是否成功", required = true)
private boolean success;
@ApiModelProperty("承载数据")
private T data;
@ApiModelProperty(value = "返回消息",required = true)
private String msg;
private R(IResultCode resultCode, T data) {
this(resultCode, data, resultCode.getMessage());
}
private R(IResultCode resultCode, T data, String msg) {
this(resultCode.getCode(), data, msg);
}
private R(int code, T data, String msg) {
this.code = code;
this.data = data;
this.msg = msg;
this.success = ResultCode.SUCCESS.code == code;
}
public static <T> R<T> data(T data) {
return data(data, "操作成功");
}
public static <T> R<T> data(T data, String msg) {
return data(200, data, msg);
}
public static <T> R<T> data(int code, T data, String msg) {
return new R(code, data, data == null ? "暂无承载数据" : msg);
}
public static <T> R<T> success(String msg) {
return new R(ResultCode.SUCCESS, msg);
}
public static <T> R<T> success(IResultCode resultCode, String msg) {
return new R(resultCode, msg);
}
public static <T> R<T> fail(String msg) {
return new R(ResultCode.FAILURE, msg);
}
public static <T> R<T> fail(int code, String msg) {
return new R(code, (Object)null, msg);
}
public static <T> R<T> fail(IResultCode resultCode, String msg) {
return new R(resultCode, msg);
}
public static <T> R<T> status(boolean flag) {
return flag ? success("操作成功") : fail("操作失败");
}
public int getCode() {
return this.code;
}
public boolean isSuccess() {
return this.success;
}
public T getData() {
return this.data;
}
public String getMsg() {
return this.msg;
}
public void setCode(final int code) {
this.code = code;
}
public void setSuccess(final boolean success) {
this.success = success;
}
public void setData(final T data) {
this.data = data;
}
public void setMsg(final String msg) {
this.msg = msg;
}
public String toString() {
return "R(code=" + this.getCode() + ", success=" + this.isSuccess() + ", data=" + this.getData() + ", msg=" + this.getMsg() + ")";
}
public R() {
}
}
public enum ResultCode implements IResultCode {
SUCCESS(200, "操作成功"),
FAILURE(400, "业务异常"),
UN_AUTHORIZED(401, "请求未授权"),
CLIENT_UN_AUTHORIZED(401, "客户端请求未授权"),
NOT_FOUND(404, "404 没找到请求"),
MSG_NOT_READABLE(400, "消息不能读取"),
METHOD_NOT_SUPPORTED(405, "不支持当前请求方法"),
MEDIA_TYPE_NOT_SUPPORTED(415, "不支持当前媒体类型"),
REQ_REJECT(403, "请求被拒绝"),
INTERNAL_SERVER_ERROR(500, "服务器异常"),
PARAM_MISS(400, "缺少必要的请求参数"),
PARAM_TYPE_ERROR(400, "请求参数类型错误"),
PARAM_BIND_ERROR(400, "请求参数绑定错误"),
PARAM_VALID_ERROR(400, "参数校验失败");
final int code;
final String message;
public int getCode() {
return this.code;
}
public String getMessage() {
return this.message;
}
private ResultCode(final int code, final String message) {
this.code = code;
this.message = message;
}
}
启动类
启动时加载FFmpegFrameGrabber和FFmpegFrameRecorder；服务结束后销毁，释放资源
@Slf4j
@SpringBootApplication
@EnableScheduling
public class FodCameraApplication {
public static void main(String[] args) {
// 服务启动执行FFmpegFrameGrabber和FFmpegFrameRecorder的tryLoad()，以免导致第一次推流时耗时。
try {
FFmpegFrameGrabber.tryLoad();
FFmpegFrameRecorder.tryLoad();
} catch (Exception e) {
e.printStackTrace();
}
SpringApplication.run(FodCameraApplication.class, args);
}
@PreDestroy
public void destory() {
log.info("服务结束，销毁...");
// 结束正在进行的任务
Set<String> keys = CameraVideoController.JOBMAP.keySet();
for (String key : keys) {
CameraVideoController.JOBMAP.get(key).setInterrupted(key);
}
// 关闭线程池
CameraThread.MyRunnable.threadPool.shutdown();
}
}
测试验证
rtmp播放
写在最后
本项目再持续优化中，博客中也已经贴出了关键代码，需要源码私信，有偿，介意勿扰！本项目设定只能在同一时间观看一个摄像头；但项目中需求可能要同时观看多个摄像头…项目持续更新中（如果放开限制会导致各个视频卡顿，应该是受公司带宽网络等问题的影响），需要的可以关注探讨。
版权声明：本文为weixin_45189427原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。