Use Java CV to implement HTTP-FLV, Hls streaming media server

preface

Generally, RTSP video streams are obtained from the camera. However, the browser cannot play RTSP video streams directly, and the streaming service needs to transcode them. A simple comparison of several protocols supported by browsers:

The RTMP protocol requires Flash support, which is now obsolete, so this option is not considered. Here we choose to implement http-FLV and HLS format streaming media conversion.

FFmpeg

FFmpeg is a video format conversion tool. Transcoding operation in the project, mainly through this tool to complete. In Java, of course. The tool-related library files are wrapped in Java CV dependencies. FFmpeg is introduced

FFmpeg is an open source computer program that can record, convert, and stream digital audio and video. Under LGPL or GPL license. It provides a complete solution for recording, converting and streaming audio and video. It includes a very advanced audio/video codec library called LibavCodec. In order to ensure high portability and codec quality, much of libavCodec code is developed from scratch.

Create a stream puller

   /** * Create a puller **@return boolean
     */
    public MediaThread createGrabber(a) {
        log.info(| | "{} {} {} | create inverter...", sourceAddress, videoType, conversionType);
        / / inverter
        grabber = new FFmpegFrameGrabber(sourceAddress);
        // Timeout (5 seconds)
        grabber.setOption("stimoout"."5000000");
        grabber.setOption("threads"."1");
        grabber.setPixelFormat(avutil.AV_PIX_FMT_YUV420P);
        // Set the cache size to improve the image quality and reduce the lag screen
        grabber.setOption("buffer_size"."1024000");
// grabber.setOption("buffer_size", "100");
        // If RTSP stream, add configuration
        if ("rtsp".equals(sourceAddress.substring(0.4))) {
            // Set the open protocol TCP/udp
            grabber.setOption("rtsp_transport"."tcp");
            // TCP is preferred for RTP transmission
            grabber.setOption("rtsp_flags"."prefer_tcp");
            // Set the timeout period
            // -stimeout is in us microseconds (1 second =1*1000*1000 microseconds).
            grabber.setOption("stimeout"."5 * 1000 * 1000");
        }
        try {
            grabber.start();
            grabberStatus = true;
        } catch (FrameGrabber.Exception e) {
            log.error("{} | create pull is abnormal!", sourceAddress, e);
            setProgress(StreamMediaConstant.SERVICE_START_PROGRESS_FAILURE, "Failed to create streamer!");
        }
        return this;
    }
Copy the code

    /**
     * flv转码
     */
    private void createATranscoderFlv(a) {

        if (StringUtils.isBlank(playAddress)) {
            // Generate the viewing address
            this.playAddress = StreamMediaUtil.generatePlaybackAddress() + ".flv";
        }
        recorder = new FFmpegFrameRecorder(flvOutputStream, grabber.getImageWidth(), grabber.getImageHeight(), grabber.getAudioChannels());
        recorder.setFormat(this.videoType);
        // Determine whether a copy is supported
        if (this.supportReuse()) {
            try {
                log.info("{} | start converting used recorder...", sourceAddress);
                recorder.start(grabber.getFormatContext());
                recorderStatus = true;
                transferFlag = true;
            } catch (FrameRecorder.Exception e) {
                log.error("{} | start converting used recorder failed!", sourceAddress, e);
                setProgress(StreamMediaConstant.SERVICE_START_PROGRESS_FAILURE, "Failed to start recorder for override!");
            }
            return;
        }


        / / code
        log.info("{} | start Flv transcoding recorder...", sourceAddress);
        recorder.setInterleaved(false);
        recorder.setVideoOption("tune"."zerolatency");
        recorder.setVideoOption("preset"."ultrafast");
        recorder.setVideoOption("crf"."26");
        recorder.setVideoOption("threads"."1");
        recorder.setFrameRate(25);// Set the frame rate
        recorder.setGopSize(25);// Set gop, same as frame rate, equivalent to chan's keyframe interval of 1 second
// recorder.setVideoBitrate(500 * 1000); / / code rate is 500 KB/s
        recorder.setVideoCodecName("libx264");
// recorder.setVideoCodec(avcodec.AV_CODEC_ID_H264);
        recorder.setPixelFormat(avutil.AV_PIX_FMT_YUV420P);
// recorder.setAudioCodec(avcodec.AV_CODEC_ID_AAC);
        recorder.setAudioCodecName("aac");
        try {
            recorder.start();
            recorderStatus = true;
        } catch (FrameRecorder.Exception e) {
            log.error("{} | create transcoding recorder exception!", sourceAddress, e);
            setProgress(StreamMediaConstant.SERVICE_START_PROGRESS_FAILURE, "Failed to create transcoding recorder!"); }}Copy the code

Hls

When transcoding in Hls mode, you need to save the generated video slice to the local disk.

  
    /** * HLS transcode */
    private void createATranscoderHls(a) {

        // Create a folder
        // Generate the viewing address
        if (StringUtils.isBlank(playAddress)) {
            playAddress = StreamMediaUtil.generatePlaybackAddress();
        }
        String path = StreamMediaConstant.HLS_DIR + "/" + playAddress + "/";
        thisVideoPath = path;
        // Check if the folder exists, if not, create it
        File file = new File(path);
        if(! file.exists()) { file.mkdir(); } path += StreamMediaConstant.VIDEO_FILE_NAME_HLS; recorder =new FFmpegFrameRecorder(path, grabber.getImageWidth(), grabber.getImageHeight(), grabber.getAudioChannels());
        recorder.setFormat("hls");

        Hls_wrap is an obsolete configuration. Ffmpeg recommends using HLS_LIST_size and HLS_FLAGS delete_segments instead of HLs_wrap
        // Set the length of time in seconds for a single TS slice. The default value is 2 seconds
        recorder.setOption("hls_time"."2");
        // Do not slice according to gop interval, force to use HLs_time to slice TS fragment
// recorder.setOption("hls_flags", "split_by_time");

        // Set the maximum number of playlist entries. If set to 0, the list file will contain all fragments, and the default is 5
        // When the time of slicing is not controlled, the number of slicing is too small, and the phenomenon of stalling will occur
        recorder.setOption("hls_list_size"."4");
        If the number of slices is greater than hlS_list_size, the previous TS slices will be automatically deleted and only hls_list_size slices will be retained
        recorder.setOption("hls_flags"."delete_segments");
        // Automatic deletion threshold for TS slices. The default value is 1, indicating that slices older than HLS_LIST_size +1 will be deleted
        recorder.setOption("hls_delete_threshold"."1");
        /* HLS slicing type: * 'MPEGTS' : outputs TS slicing files in MPEG-2 transport stream format, compatible with all HLS versions. * 'FMP4 ': Sliced files are Fragmented MP4 format, similar to MPEG-DASH. Fmp4 files are available for HLS Version 7 and higher. * /
        recorder.setOption("hls_segment_type"."mpegts");
        File000. ts, file001.ts, file002.ts, and so on will be generated
// recorder.setOption("hls_segment_filename", path + "-%03d.ts");
        recorder.setOption("hls_segment_filename", path + "-%5d.ts");
        // Set the number of the first slice

        recorder.setOption("start_number", String.valueOf(tsCont));
        recorder.setPixelFormat(avutil.AV_PIX_FMT_YUV420P);

        // Determine whether a copy is supported
        if (this.supportReuse()) {
            try {
                log.info("{} | start Hls converting used recorder...", sourceAddress);
                recorder.start(grabber.getFormatContext());
                recorderStatus = true;
                transferFlag = true;
            } catch (FrameRecorder.Exception e) {
                log.error("{} | start converting used recorder failed!", sourceAddress, e);
                setProgress(StreamMediaConstant.SERVICE_START_PROGRESS_FAILURE, "Failed to start recorder for override!");
            }
            return;
        }


        / / code
        log.info("{} | start Hls transcoding recorder...", sourceAddress);
        // Set zero latency
        recorder.setVideoOption("tune"."zerolatency");
        / / fast
        recorder.setVideoOption("preset"."ultrafast");
// recorder.setVideoOption("crf", "26");
// recorder.setVideoOption("threads", "1");
        recorder.setFrameRate(25);// Set the frame rate
        recorder.setGopSize(25);// Set gop, same as frame rate, equivalent to chan's keyframe interval of 1 second
// recorder.setVideoBitrate(500 * 1000); / / code rate is 500 KB/s
        recorder.setVideoCodec(avcodec.AV_CODEC_ID_H264);
        recorder.setPixelFormat(avutil.AV_PIX_FMT_YUV420P);
        recorder.setAudioCodec(avcodec.AV_CODEC_ID_AAC);
        try {
            recorder.start();
            recorderStatus = true;
        } catch (FrameRecorder.Exception e) {
            log.error("{} | create transcoding recorder exception!", sourceAddress, e);
            setProgress(StreamMediaConstant.SERVICE_START_PROGRESS_FAILURE, "Failed to create transcoding recorder!"); }}Copy the code

Hls

    /** * Enable the forwarding service ** /
    public void transform(a) {
        if(! grabberStatus || ! recorderStatus) {return;
        }
        log.info("Turn {} | open flow operation...", sourceAddress);
        try {
            grabber.flush();
        } catch (FrameGrabber.Exception e) {
            log.error("{} | empty pull current limiter cache failed!", sourceAddress, e);
        }
        String mapKey = sourceAddress + "-" + videoType;
        setProgress(StreamMediaConstant.SERVICE_START_PROGRESS_SUCCESS, "");
        StreamMediaConstant.MEDIA_INFO_MAP.put(mapKey, this);
        // The service is started
        starting = false;


        // Time stamp calculation
        int errorCount = 0;
        long currentTimeMillis;
        running = true;
        while (running) {
            currentTimeMillis = System.currentTimeMillis();

            // If there are too many errors, re-establish the connection or restart
            if (errorCount > 10 || reConnection) {
                if (transferFlag) {
                    // Re-establishing the connection does not take effect for remultiplexing, so restart the service
                    restartService = true;
                    break;
                }
                try {
                    log.info("{} | to establish links to...", sourceAddress);
                    grabber.restart();
                    grabber.flush();
                    // Reset the setup successfully and clear the error count to zero
                    errorCount = 0;
                } catch (FrameGrabber.Exception e1) {
                    log.error("{} | to establish a connection failure", sourceAddress, e1); errorCount++; }}// Turn the current
            try {
                if (transferFlag) {
                    / / converting
                    AVPacket pkt = grabber.grabPacket();
                    if (null == pkt || pkt.isNull()) {
                        log.error("{} | pkt is null", sourceAddress);
                        errorCount++;
                        continue;
                    }
                    recorder.recordPacket(pkt);
                    av_packet_unref(pkt);
                } else {
                    / / code
                    Frame frame = grabber.grabFrame();
                    if (frame == null) {
                        log.error("{} | frame is null", sourceAddress);
                        errorCount++;
                        continue; } recorder.record(frame); }}catch (Exception e) {
                log.error("{} | turn flow operation exception!", sourceAddress, e);
                errorCount++;
            }

            // Check whether the HLS connection is alive. This command is executed every 3 minutes
            if (currentTimeMillis - httpCheckTime > (1000 * 60 * 3)) { httpCheckTime = currentTimeMillis; checkHlsHttp(); }}// Close contains stop and release methods. Recording files must ensure that the stop() method is executed at the end
        try {
            recorder.close();
            grabber.close();
            if (null != tscCache) {
                tscCache.clear();
            }
        } catch (IOException e) {
            log.error(End of turn "{} | flow operation, shut off the flow anomalies!", sourceAddress, e);
        }

        // Delete local files
        File file = new File(thisVideoPath);
        if (file.exists()) {
            log.info(End of turn "{} | flow, delete local files.", sourceAddress);
            FileUtil.del(file);
        }

        // A restart is required
        if (restartService) {
            // Restart only once a minute
            log.info("{} | restart...", sourceAddress);
            restartService = false;
            this.grabberStatus = false;
            this.recorderStatus = false;
            starting = true;
            reConnection = false;
            lastRestartTime = System.currentTimeMillis();
            this.createGrabber().createRecodeRecorder().transform();
        }

        log.info(End of turn "{} | flow operation", sourceAddress);
        StreamMediaConstant.MEDIA_INFO_MAP.remove(mapKey);
    }

Copy the code

 /** * Enable the forwarding service **@return view
     */
    public void transform(a) {
        if(! grabberStatus || ! recorderStatus) {return;
        }
        log.info("{} | open flow service...", sourceAddress);
        try {
            grabber.flush();
        } catch (FrameGrabber.Exception e) {
            log.error("{} | empty pull current limiter cache failure", sourceAddress, e);
        }
        if (flvHeader == null) {
            flvHeader = flvOutputStream.toByteArray();
            flvOutputStream.reset();
        }
        String mapKey = sourceAddress + "-" + videoType;
        setProgress(StreamMediaConstant.SERVICE_START_PROGRESS_SUCCESS, "");
        StreamMediaConstant.MEDIA_INFO_MAP.put(mapKey, this);
        // The service is started
        starting = false;


        // Time stamp calculation
        int errorCount = 0;
        running = true;
        long videoTS;
        while (running) {
            // If there are too many errors, re-establish the connection or restart
            if (errorCount > 10) {
                if (transferFlag) {
                    // Transcoding to re-establish the connection does not take effect, so restart the service
                    restartService = true;
                    break;
                }
                try {
                    log.info("{} | to establish links to...", sourceAddress);
                    grabber.restart();
                    grabber.flush();
                    // Reset the setup successfully and clear the error count to zero
                    errorCount = 0;
                } catch (FrameGrabber.Exception e1) {
                    log.error("{} | to establish a connection failure", sourceAddress, e1); errorCount++; }}try {

                if (transferFlag) {
                    
                    / / converting
                    AVPacket pkt = grabber.grabPacket();
                    if (null == pkt || pkt.isNull()) {
                        log.error("{} | pkt is null", sourceAddress);
                        errorCount++;
                        continue;
                    }
                    recorder.recordPacket(pkt);
                    av_packet_unref(pkt);
                } else {
                    / / code
                    Frame frame = grabber.grabFrame();
                    if (frame == null) {
                        log.error("{} | frame is null", sourceAddress);
                        errorCount++;
                        continue; } recorder.record(frame); }}catch (Exception e) {
                log.error("{} | to turn abnormal operation flow", sourceAddress, e);
                errorCount++;
            }


            if (flvOutputStream.size() > 0) {
                theLatestData = flvOutputStream.toByteArray();
                // Send the video to the front-endsendFlvFrameData(theLatestData); flvOutputStream.reset(); }}// Close contains stop and release methods. Recording files must ensure that the stop() method is executed at the end
        try {
            recorder.close();
            grabber.close();
            flvOutputStream.close();
        } catch (IOException e) {
            log.error(End of turn "{} | flow operation, shut off the flow.", sourceAddress, e);
        }

        // Restart the system
        if (restartService) {
            log.info("{} | restart...", sourceAddress);
            flvOutputStream = new ByteArrayOutputStream();
            restartService = false;
            this.grabberStatus = false;
            this.recorderStatus = false;
            starting = true;
            firstPushAfterReStart = true;
            this.createGrabber().createRecodeRecorder().transform();
        }
        log.info("{} | end streaming service.", sourceAddress);
        StreamMediaConstant.MEDIA_INFO_MAP.remove(mapKey);
    }
Copy the code

Listen for Hls requests

The essence of Hls video stream is to cut the video into small video clips, and then the client obtains these video clips through HTTP request for playback. Therefore, to listen to Hls requests, you only need to read files from disks according to request parameters.

    public void getHlsFile(HttpServletRequest request, HttpServletResponse response, String playName) {

        String ipAddress = StreamMediaUtil.getIpAddress(request);
        response.addHeader("Content-Disposition"."attachment; filename=play.m3u8");
        response.setHeader("Access-Control-Allow-Origin"."*");
        if (playName.toLowerCase().endsWith("m3u8")) {
            response.setContentType("application/x-mpegURL");
        } else {
            response.setContentType("video/MP2T");
        }
        /* .M3U8 application/x-mpegURL or vnd.apple.mpegURL .ts video/MP2T */
        if (null == httpMap.get(ipAddress)) {
            log.info("{} | {} new connection! Current connections: {}", sourceAddress, ipAddress, httHlsClientMap.size() + 1);
        }
        httHlsClientMap.put(ipAddress, System.currentTimeMillis());
        httpMap.put(ipAddress, System.currentTimeMillis());

        String pathFile = thisVideoPath + playName;
        byte[] fileByte = getFileByte(pathFile);
        if (null == fileByte) {
            return;
        }
        ServletOutputStream outputStream = null;
        try {
            outputStream = response.getOutputStream();
            outputStream.write(fileByte);
            outputStream.flush();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (null! = outputStream) {try {
                    outputStream.close();
                } catch(IOException e) { e.printStackTrace(); }}}}/** * Get file data **@paramPathFile File address *@return* /
    private byte[] getFileByte(String pathFile) {
        // The service is being restarted
        if (starting || restartService) {
            while (true) {
                if (starting || restartService) {
                    try {
                        Thread.sleep(1000);
                    } catch(InterruptedException e) { e.printStackTrace(); }}else {
                    break;
                }
            }
        }

        Object bites = null;
        if (pathFile.endsWith(".ts")) {
            // Get the latest ts subscript
            tsCont = Integer.parseInt(pathFile.substring(pathFile.lastIndexOf("-") + 1, pathFile.lastIndexOf(".")));
            if (this.tscCache.containsKey(pathFile)) { bites = tscCache.get(pathFile); }}if (null! = bites) {return (byte[]) bites;
        }
        File file = new File(pathFile);
        if(! file.exists()) {return null;
        }

        FileReader fileReader = FileReader.create(file);
        byte[] bytes = fileReader.readBytes();
        // It is possible to cache some data to reduce disk file reading speed when multiple clients play simultaneously. When the cache is full, the first objects are cleared
        tscCache.put(pathFile, bytes, (1000 * 6));
        return bytes;
    }

Copy the code

 /** * Add HTTP client */
    public void addFlvHttpClient(HttpServletRequest request, HttpServletResponse response) {
        if (running) {
            response.addHeader("Content-Disposition"."attachment; filename=\"" + playAddress + "\" ");
            response.setContentType("video/x-flv");
            response.setHeader("Connection"."keep-alive");
            response.setHeader("accept_ranges"."bytes");
            response.setHeader("pragma"."no-cache");
            response.setHeader("cache_control"."no-cache");
            response.setHeader("transfer_encoding"."CHUNKED");
            response.setHeader("SERVER"."hmsm");
            String ipAddress = StreamMediaUtil.getIpAddress(request);


            response.setStatus(200);
            ServletOutputStream outputStream = null;
            try {
                outputStream = response.getOutputStream();
            } catch (IOException e) {
                log.error("{} | {} outputStream for failure!", sourceAddress, ipAddress, e);
            }
            if (null == outputStream) {
                return;
            }
            try {
                outputStream.write(flvHeader, 0, flvHeader.length);
                outputStream.flush();
            } catch (IOException e) {
                log.error("{} | {} writes data failed head!", sourceAddress, ipAddress, e);
            }

            httpFlvClientMap.put(ipAddress, outputStream);
            httpMap.put(ipAddress, System.currentTimeMillis());
            log.info("{} | {} new connection! Current connections: {}", sourceAddress, ipAddress, httpMap.size());
            while (true) {
                // The thread is not released, and the connection remains.
                if (null == httpFlvClientMap.get(ipAddress)) {
                    // When the link is removed, the method is terminated and the thread is closed.
                    log.error("{} | {} end thread!", sourceAddress, ipAddress);
                    break;
                }
                try {
                    // Thread sleep, the HTTP connection will not be broken as long as the thread is not terminated
                    // Don't know if there is a better way
                    Thread.sleep(1000 * 60 * 1);
                } catch(InterruptedException e) { e.printStackTrace(); }}}}private void sendFlvFrameData(byte[] data) {
        Iterator<Map.Entry<String, ServletOutputStream>> iterator = httpFlvClientMap.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<String, ServletOutputStream> next = iterator.next();
            ServletOutputStream outputStream = null;
            try {
                outputStream = next.getValue();
                outputStream.write(data, 0, data.length);
                outputStream.flush();
            } catch (java.lang.Exception e) {
                try {
                    if (null != outputStream) {
                        outputStream.close();
                    }
                } catch (IOException ioException) {
                    ioException.printStackTrace();
                }
                httpMap.remove(next.getKey());
                iterator.remove();
                log.error("{} | {} | push flow failures, close the connection. Remaining connections: {}", sourceAddress, next.getKey(), httpMap.size(), e);
            }
        }
        firstPushAfterReStart = false;
        hasClient();
    }

Copy the code

Matters needing attention

1. When performed in a way that is long term, can occurnull == pktIs currently suspected to be andbuffer_size No solution was found. Restart the service to deal with it.
2. In Hls mode, the number of slices and slice time should not be too small, otherwise there will be delay and lag. Of course, this depends on server performance and network conditions.
3. The test found that the delay of Flv form is about 3~5 seconds, and the delay of Hls form is about 6 seconds. Flv also performs better than Hls.

The source address

Gitee.com/yefengr/ast…

reference

HLS,HTTP,RTSP,RTMP protocol differences JavaCV development practice tutorial (JavaCV tutorial) code a lot of reference: EasyMedia