java 实现监控rtsp流转flv,实现前端播放(前后端代码都有)

一.服务代码

1. 目录结构

2. maven配置文件引入坐标:

org.bytedecojavacv-platform1.5.1javax.xml.bindjaxb-api2.3.0

3. 服务器代码

controller层:

import com.xr.web.rtspconverterflvspringbootstarter.service.IFLVService;import io.swagger.annotations.Api;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.web.bind.annotation.GetMapping;import org.springframework.web.bind.annotation.RequestMapping;import org.springframework.web.bind.annotation.RestController;import javax.servlet.http.HttpServletRequest;import javax.servlet.http.HttpServletResponse;/** * FLV流转换 * * @author gc.x */@Api(tags = "flv")@RequestMapping("/flv")@RestControllerpublic class FLVController {@Autowiredprivate IFLVService service;@GetMapping()public void open4(HttpServletResponse response,HttpServletRequest request) {String test = "rtsp://admin:sdxr@2022@192.168.0.205:554";service.open(test, response, request);}}

config层:

import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.scheduling.TaskScheduler;import org.springframework.scheduling.annotation.EnableScheduling;import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;/** * 使用多线程执行定时任务 ** @author gc.x * */@Configuration@EnableSchedulingpublic class SchedulerConfig {@Beanpublic TaskScheduler taskScheduler() {ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();// 线程池大小scheduler.setPoolSize(3);// 线程名字前缀scheduler.setThreadNamePrefix("task-thread-");return scheduler;}}

factories层:

/** * 转换器状态（初始化、打开、关闭、错误、运行） ** @author gc.x */public enum ConverterState {INITIAL, OPEN, CLOSE, ERROR, RUN}

import javax.servlet.AsyncContext;import java.io.IOException;public interface Converter {/** * 获取该转换的key */public String getKey();/** * 获取该转换的url ** @return */public String getUrl();/** * 添加一个流输出 ** @param entity */public void addOutputStreamEntity(String key, AsyncContext entity) throws IOException;/** * 退出转换 */public void exit();/** * 启动 */public void start();}

import com.alibaba.fastjson.util.IOUtils;import lombok.extern.slf4j.Slf4j;import org.bytedeco.ffmpeg.avcodec.AVPacket;import org.bytedeco.ffmpeg.global.avcodec;import org.bytedeco.javacv.FFmpegFrameGrabber;import org.bytedeco.javacv.FFmpegFrameRecorder;import javax.servlet.AsyncContext;import java.io.ByteArrayOutputStream;import java.io.IOException;import java.util.Iterator;import java.util.List;import java.util.Map;/** * javacv转包装
 * 无须转码，更低的资源消耗，更低的延迟
 * 确保流来源视频H264格式,音频AAC格式 ** @author gc.x */@Slf4jpublic class ConverterFactories extends Thread implements Converter {public volatile boolean runing = true;/** * 读流器 */private FFmpegFrameGrabber grabber;/** * 转码器 */private FFmpegFrameRecorder recorder;/** * 转FLV格式的头信息
 * 如果有第二个客户端播放首先要返回头信息 */private byte[] headers;/** * 保存转换好的流 */private ByteArrayOutputStream stream;/** * 流地址，h264,aac */private String url;/** * 流输出 */private List outEntitys;/** * key用于表示这个转换器 */private String key;/** * 转换队列 */private Map factories;public ConverterFactories(String url, String key, Map factories, List outEntitys) {this.url = url;this.key = key;this.factories = factories;this.outEntitys = outEntitys;}@Overridepublic void run() {boolean isCloseGrabberAndResponse = true;try {grabber = new FFmpegFrameGrabber(url);if ("rtsp".equals(url.substring(0, 4))) {grabber.setOption("rtsp_transport", "tcp");grabber.setOption("stimeout", "5000000");}grabber.start();if (avcodec.AV_CODEC_ID_H264 == grabber.getVideoCodec()&& (grabber.getAudioChannels() == 0 || avcodec.AV_CODEC_ID_AAC == grabber.getAudioCodec())) {log.info("this url:{} converterFactories start", url);// 来源视频H264格式,音频AAC格式// 无须转码，更低的资源消耗，更低的延迟stream = new ByteArrayOutputStream();recorder = new FFmpegFrameRecorder(stream, grabber.getImageWidth(), grabber.getImageHeight(),grabber.getAudioChannels());recorder.setInterleaved(true);recorder.setVideoOption("preset", "ultrafast");recorder.setVideoOption("tune", "zerolatency");recorder.setVideoOption("crf", "25");recorder.setFrameRate(grabber.getFrameRate());recorder.setSampleRate(grabber.getSampleRate());if (grabber.getAudioChannels() > 0) {recorder.setAudioChannels(grabber.getAudioChannels());recorder.setAudioBitrate(grabber.getAudioBitrate());recorder.setAudioCodec(grabber.getAudioCodec());}recorder.setFormat("flv");recorder.setVideoBitrate(grabber.getVideoBitrate());recorder.setVideoCodec(grabber.getVideoCodec());recorder.start(grabber.getFormatContext());if (headers == null) {headers = stream.toByteArray();stream.reset();writeResponse(headers);}int nullNumber = 0;while (runing) {AVPacket k = grabber.grabPacket();if (k != null) {try {recorder.recordPacket(k);} catch (Exception e) {}if (stream.size() > 0) {byte[] b = stream.toByteArray();stream.reset();writeResponse(b);if (outEntitys.isEmpty()) {log.info("没有输出退出");break;}}avcodec.av_packet_unref(k);} else {nullNumber++;if (nullNumber > 200) {break;}}Thread.sleep(5);}} else {isCloseGrabberAndResponse = false;// 需要转码为视频H264格式,音频AAC格式ConverterTranFactories c = new ConverterTranFactories(url, key, factories, outEntitys, grabber);factories.put(key, c);c.start();}} catch (Exception e) {log.error(e.getMessage(), e);} finally {closeConverter(isCloseGrabberAndResponse);completeResponse(isCloseGrabberAndResponse);log.info("this url:{} converterFactories exit", url);}}/** * 输出FLV视频流 ** @param b */public void writeResponse(byte[] b) {Iterator it = outEntitys.iterator();while (it.hasNext()) {AsyncContext o = it.next();try {o.getResponse().getOutputStream().write(b);} catch (Exception e) {log.info("移除一个输出");it.remove();}}}/** * 退出转换 */public void closeConverter(boolean isCloseGrabberAndResponse) {if (isCloseGrabberAndResponse) {IOUtils.close(grabber);factories.remove(this.key);}IOUtils.close(recorder);IOUtils.close(stream);}/** * 关闭异步响应 ** @param isCloseGrabberAndResponse */public void completeResponse(boolean isCloseGrabberAndResponse) {if (isCloseGrabberAndResponse) {Iterator it = outEntitys.iterator();while (it.hasNext()) {AsyncContext o = it.next();o.complete();}}}@Overridepublic String getKey() {return this.key;}@Overridepublic String getUrl() {return this.url;}@Overridepublic void addOutputStreamEntity(String key, AsyncContext entity) throws IOException {if (headers == null) {outEntitys.add(entity);} else {entity.getResponse().getOutputStream().write(headers);entity.getResponse().getOutputStream().flush();outEntitys.add(entity);}}@Overridepublic void exit() {this.runing = false;try {this.join();} catch (Exception e) {log.error(e.getMessage(), e);}}}

import com.alibaba.fastjson.util.IOUtils;import lombok.extern.slf4j.Slf4j;import org.bytedeco.ffmpeg.global.avcodec;import org.bytedeco.javacv.FFmpegFrameGrabber;import org.bytedeco.javacv.FFmpegFrameRecorder;import org.bytedeco.javacv.Frame;import javax.servlet.AsyncContext;import java.io.ByteArrayOutputStream;import java.io.IOException;import java.util.Iterator;import java.util.List;import java.util.Map;/** * javacv转码
 * 流来源不是视频H264格式,音频AAC格式 转码为视频H264格式,音频AAC格式 ** @author gc.x */@Slf4jpublic class ConverterTranFactories extends Thread implements Converter {public volatile boolean runing = true;/** * 读流器 */private FFmpegFrameGrabber grabber;/** * 转码器 */private FFmpegFrameRecorder recorder;/** * 转FLV格式的头信息
 * 如果有第二个客户端播放首先要返回头信息 */private byte[] headers;/** * 保存转换好的流 */private ByteArrayOutputStream stream;/** * 流地址，h264,aac */private String url;/** * 流输出 */private List outEntitys;/** * key用于表示这个转换器 */private String key;/** * 转换队列 */private Map factories;public ConverterTranFactories(String url, String key, Map factories,List outEntitys, FFmpegFrameGrabber grabber) {this.url = url;this.key = key;this.factories = factories;this.outEntitys = outEntitys;this.grabber = grabber;}@Overridepublic void run() {try {log.info("this url:{} converterTranFactories start", url);grabber.setFrameRate(25);if (grabber.getImageWidth() > 1920) {grabber.setImageWidth(1920);}if (grabber.getImageHeight() > 1080) {grabber.setImageHeight(1080);}stream = new ByteArrayOutputStream();recorder = new FFmpegFrameRecorder(stream, grabber.getImageWidth(), grabber.getImageHeight(),grabber.getAudioChannels());recorder.setInterleaved(true);recorder.setVideoOption("preset", "ultrafast");recorder.setVideoOption("tune", "zerolatency");recorder.setVideoOption("crf", "25");recorder.setGopSize(50);recorder.setFrameRate(25);recorder.setSampleRate(grabber.getSampleRate());if (grabber.getAudioChannels() > 0) {recorder.setAudioChannels(grabber.getAudioChannels());recorder.setAudioBitrate(grabber.getAudioBitrate());recorder.setAudioCodec(avcodec.AV_CODEC_ID_AAC);}recorder.setFormat("flv");recorder.setVideoBitrate(grabber.getVideoBitrate());recorder.setVideoCodec(avcodec.AV_CODEC_ID_H264);recorder.start();if (headers == null) {headers = stream.toByteArray();stream.reset();writeResponse(headers);}int nullNumber = 0;while (runing) {// 抓取一帧Frame f = grabber.grab();if (f != null) {try {// 转码recorder.record(f);} catch (Exception e) {}if (stream.size() > 0) {byte[] b = stream.toByteArray();stream.reset();writeResponse(b);if (outEntitys.isEmpty()) {log.info("没有输出退出");break;}}} else {nullNumber++;if (nullNumber > 200) {break;}}Thread.sleep(5);}} catch (Exception e) {log.error(e.getMessage(), e);} finally {closeConverter();completeResponse();log.info("this url:{} converterTranFactories exit", url);factories.remove(this.key);}}/** * 输出FLV视频流 ** @param b */public void writeResponse(byte[] b) {Iterator it = outEntitys.iterator();while (it.hasNext()) {AsyncContext o = it.next();try {o.getResponse().getOutputStream().write(b);} catch (Exception e) {log.info("移除一个输出");it.remove();}}}/** * 退出转换 */public void closeConverter() {IOUtils.close(grabber);IOUtils.close(recorder);IOUtils.close(stream);}/** * 关闭异步响应 */public void completeResponse() {Iterator it = outEntitys.iterator();while (it.hasNext()) {AsyncContext o = it.next();o.complete();}}@Overridepublic String getKey() {return this.key;}@Overridepublic String getUrl() {return this.url;}@Overridepublic void addOutputStreamEntity(String key, AsyncContext entity) throws IOException {if (headers == null) {outEntitys.add(entity);} else {entity.getResponse().getOutputStream().write(headers);entity.getResponse().getOutputStream().flush();outEntitys.add(entity);}}@Overridepublic void exit() {this.runing = false;try {this.join();} catch (Exception e) {log.error(e.getMessage(), e);}}}

result层:

import com.alibaba.fastjson.JSONObject;import java.io.Serializable;import java.util.HashMap;/** * 封装返回结果 **/public class JsonResult extends HashMap implements Serializable {private static final long serialVersionUID = 1L;public static final int SUCCESS = 200;public JsonResult() {}/** * 返回成功 */public static JsonResult ok() {return ok("操作成功");}/** * 返回成功 */public static JsonResult okFallBack() {return okFallBack("操作成功");}/** * 返回成功 */public JsonResult put(Object obj) {return this.put("data", obj);}/** * 返回成功 */public static JsonResult ok(String message) {return result(200, message);}/** * 降级函数 - 返回成功 */public static JsonResult okFallBack(String message) {return result(205, message);}/** * 返回成功 */public static JsonResult result(int code, String message) {JsonResult jsonResult = new JsonResult();jsonResult.put("timestamp", System.currentTimeMillis());jsonResult.put("status", code);jsonResult.put("message", message);return jsonResult;}/** * 返回失败 */public static JsonResult error() {return error("操作失败");}/** * 返回失败 */public static JsonResult error(String message) {return error(500, message);}/** * 返回失败 */public static JsonResult error(int code, String message) {JsonResult jsonResult = new JsonResult();jsonResult.put("timestamp", System.currentTimeMillis());jsonResult.put("status", code);jsonResult.put("message", message);return jsonResult;}/** * 设置code */public JsonResult setCode(int code) {super.put("status", code);return this;}/** * 设置message */public JsonResult setMessage(String message) {super.put("message", message);return this;}/** * 放入object */@Overridepublic JsonResult put(String key, Object object) {super.put(key, object);return this;}/** * 权限禁止 */public static JsonResult forbidden(String message) {JsonResult jsonResult = new JsonResult();jsonResult.put("timestamp", System.currentTimeMillis());jsonResult.put("status", 401);jsonResult.put("message", message);return jsonResult;}@Overridepublic String toString() {return JSONObject.toJSONString(this);}public JSONObject toJSONObject() {return JSONObject.parseObject(toString());}}

service层:

import com.alibaba.fastjson.JSONObject;import java.io.Serializable;import java.util.HashMap;/** * 封装返回结果 **/public class JsonResult extends HashMap implements Serializable {private static final long serialVersionUID = 1L;public static final int SUCCESS = 200;public JsonResult() {}/** * 返回成功 */public static JsonResult ok() {return ok("操作成功");}/** * 返回成功 */public static JsonResult okFallBack() {return okFallBack("操作成功");}/** * 返回成功 */public JsonResult put(Object obj) {return this.put("data", obj);}/** * 返回成功 */public static JsonResult ok(String message) {return result(200, message);}/** * 降级函数 - 返回成功 */public static JsonResult okFallBack(String message) {return result(205, message);}/** * 返回成功 */public static JsonResult result(int code, String message) {JsonResult jsonResult = new JsonResult();jsonResult.put("timestamp", System.currentTimeMillis());jsonResult.put("status", code);jsonResult.put("message", message);return jsonResult;}/** * 返回失败 */public static JsonResult error() {return error("操作失败");}/** * 返回失败 */public static JsonResult error(String message) {return error(500, message);}/** * 返回失败 */public static JsonResult error(int code, String message) {JsonResult jsonResult = new JsonResult();jsonResult.put("timestamp", System.currentTimeMillis());jsonResult.put("status", code);jsonResult.put("message", message);return jsonResult;}/** * 设置code */public JsonResult setCode(int code) {super.put("status", code);return this;}/** * 设置message */public JsonResult setMessage(String message) {super.put("message", message);return this;}/** * 放入object */@Overridepublic JsonResult put(String key, Object object) {super.put(key, object);return this;}/** * 权限禁止 */public static JsonResult forbidden(String message) {JsonResult jsonResult = new JsonResult();jsonResult.put("timestamp", System.currentTimeMillis());jsonResult.put("status", 401);jsonResult.put("message", message);return jsonResult;}@Overridepublic String toString() {return JSONObject.toJSONString(this);}public JSONObject toJSONObject() {return JSONObject.parseObject(toString());}}

import javax.servlet.http.HttpServletRequest;import javax.servlet.http.HttpServletResponse;public interface IFLVService {/** * 打开一个流地址 ** @param url * @param response */public void open(String url, HttpServletResponse response, HttpServletRequest request);}

二.客户端代码基于flv.js进行播放

点击播放if (flvjs.isSupported()) {var videoElement = document.getElementById('videoElement');var flvPlayer = flvjs.createPlayer({type: 'flv',url: 'http://127.0.0.1:10010/xr/flv',});flvPlayer.attachMediaElement(videoElement);flvPlayer.load();}//自动播放,浏览器不支持function playflv() {flvPlayer.play();}

这里因为浏览器把自动播放给禁止了，加了个按钮点击事件

https://www.bootcdn.cn/

引入的flv.js文件在如下网站下载即可: