基于JavaCV对RTSP、RTMP视频流检测

1. Maven依赖

1.1 懒人版

<dependency>
      <groupId>org.bytedeco</groupId>
      <artifactId>javacv-platform</artifactId>
      <version>1.5.5</version>
</dependency>

1.2 精确版

由于整个依赖较大（约900M），以下为实际需要的依赖（不同的操作系统需要根据自己的操作系统进行适配），以下xml配置了properties及dependencies两个节点。

bytecode-sys配置根据运行系统可以自行变更为,以下部分系统值：

<properties>
    <bytecode-sys>linux-x86_64</bytecode-sys>
</properties>

<dependencies>
    <dependency>
      <groupId>org.bytedeco</groupId>
      <artifactId>ffmpeg</artifactId>
      <version>4.3.2-1.5.5</version>
      <exclusions>
        <exclusion>
          <groupId>*</groupId>
          <artifactId>*</artifactId>
        </exclusion>
      </exclusions>
    </dependency>

    <dependency>
      <groupId>org.bytedeco</groupId>
      <artifactId>ffmpeg</artifactId>
      <classifier>${bytecode-sys}</classifier>
      <version>4.3.2-1.5.5</version>
      <exclusions>
        <exclusion>
          <groupId>*</groupId>
          <artifactId>*</artifactId>
        </exclusion>
      </exclusions>
    </dependency>
    <dependency>
      <groupId>org.bytedeco</groupId>
      <artifactId>javacv</artifactId>
      <version>1.5.5</version>
      <exclusions>
        <exclusion>
          <groupId>*</groupId>
          <artifactId>*</artifactId>
        </exclusion>
      </exclusions>
    </dependency>
    <dependency>
      <groupId>org.bytedeco</groupId>
      <artifactId>javacpp</artifactId>
      <version>1.5.5</version>
      <exclusions>
        <exclusion>
          <groupId>*</groupId>
          <artifactId>*</artifactId>
        </exclusion>
      </exclusions>
    </dependency>
    <dependency>
      <groupId>org.bytedeco</groupId>
      <artifactId>javacpp</artifactId>
      <classifier>${bytecode-sys}</classifier>
      <version>1.5.5</version>
      <exclusions>
        <exclusion>
          <groupId>*</groupId>
          <artifactId>*</artifactId>
        </exclusion>
      </exclusions>
    </dependency>
    <!--endregion-->

  </dependencies>

2. 代码

FFmpegFrameGrabber通过grab方法可以获取视频信息，当frame不为空则正常；

但是该方法会导致阻塞，因此此处通过Future的get方法，设置一个超时时间，超时后返回false。

2.1 核心测试类

import com.hz.utils.CustomThreadPool;
import org.bytedeco.javacv.FFmpegFrameGrabber;
import org.bytedeco.javacv.Frame;
import org.junit.jupiter.api.Test;

import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;


public class RtmpCheckTest {

    @Test
    public void test01() {
        try (CustomThreadPool pool = CustomThreadPool.initPool(1, 1, 1, "ceshi");) {

            Future<Boolean> future = pool.submit(() -> {

                String rtmpUrl = "rtmp://yourRtmpServer:1935/live/test";
                FFmpegFrameGrabber grabber = new FFmpegFrameGrabber(rtmpUrl);
                try {

                    grabber.start();
                    Frame frame = grabber.grab();
                    boolean status = frame != null;
                    return status;
                } catch (Exception e) {
                    return false;
                } finally {
                    grabber.stop();
                }
            });

            Boolean status = false;
            try {
                status = future.get(10, TimeUnit.SECONDS);
            } catch (Exception e) {
                System.out.println(e.getMessage());
            }

            System.out.println("视频状态：" + status);

        }

    }

}

2.2 依赖（自定义线程池工具类）

package com.hz.utils;

import java.io.Closeable;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @description:
 * @author: pp_lan
 * @date: 2022/3/15
 */
public class CustomThreadPool implements Closeable {

    private ThreadPoolExecutor pool;

    private AtomicInteger count = new AtomicInteger();

    private CustomThreadPool(int core, int max, int queueSize, String threadName) {
        this.pool = new ThreadPoolExecutor(core, max, 3, TimeUnit.SECONDS, new ArrayBlockingQueue<>(queueSize),
                r -> new Thread(r, threadName + count.getAndIncrement()));
    }

    public ThreadPoolExecutor getPool() {
        return pool;
    }

    /**
     * 获取对象
     * @param core
     * @param max
     * @param queueSize
     * @param threadName
     * @return
     */
    public static CustomThreadPool initPool(int core, int max, int queueSize, String threadName) {
        return new CustomThreadPool(core, max, queueSize, threadName);
    }

    /**
     * callable批量任务
     * @param tasks
     * @param <T>
     * @return
     */
    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) {
        try {
            List<Future<T>> futures = this.pool.invokeAll(tasks);
            return futures;
        } catch (InterruptedException e) {
            throw new RuntimeException("线程被中断", e);
        }
    }

    /**
     * callable单个任务
     * @param task
     * @param <T>
     * @return
     */
    public <T> Future<T> submit(Callable<T> task) {
        return this.pool.submit(task);
    }

    /**
     * runnable任务
     * @param runnable
     */
    public void execute(Runnable runnable) {
        this.pool.execute(runnable);
    }

    /**
     * runnable任务
     * @param runnable
     */
    public Future<?> submit(Runnable runnable) {
        return this.pool.submit(runnable);
    }

    /**
     * 线程池关闭，建议使用try-resource进行自动关闭，不要手动执行
     */
    public void shutdown() {
        this.pool.shutdown();
        try {
            if (!this.pool.awaitTermination(1, TimeUnit.MINUTES)) {
                this.pool.shutdownNow();
                if (!this.pool.awaitTermination(1, TimeUnit.MINUTES)) {
                    throw new RuntimeException("线程池关闭失败");
                }
            }
        } catch (InterruptedException e) {
            throw new RuntimeException("线程池关闭失败", e);
        }
    }

    @Override
    public void close() {
        this.shutdown();
    }
}

3. 环境

RTSP环境搭建

4. 执行结果