专栏首页eguid开源技术分享搭建rtmp直播流服务之3:java开发ffmpeg实现rtsp转rtmp并实现ffmpeg命令的接口化管理架构设计及代码实现

搭建rtmp直播流服务之3:java开发ffmpeg实现rtsp转rtmp并实现ffmpeg命令的接口化管理架构设计及代码实现

版权声明:本文为博主原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。

本文链接:https://blog.csdn.net/eguid_1/article/details/51787646

上一篇文章简单介绍了java如何调用ffmpeg的命令:http://blog.csdn.net/eguid_1/article/details/51777716

上上一篇介绍了nginx-rtmp服务器的搭建:http://blog.csdn.net/eguid_1/article/details/51749830

这一篇将进一步深挖java对ffmepg命令的控制并最终实现服务接口化

本篇文章源码:http://download.csdn.net/detail/eguid_1/9563637

github项目地址:https://github.com/eguid/FFCH4J

通知:由于很多同学反映本章代码的命令封装设计的不是很好,所以对本章代码重新进行了实现,新版本推翻了本章原有代码内部实现,接口设计更加利于注入自己的实现,并增加可执行原生ffmpeg命令功能

新版本请到这里查看:java封装FFmpeg命令,支持原生ffmpeg全部命令,实现FFmpeg多进程处理与多线程输出控制(开启、关闭、查询),rtsp/rtmp推流、拉流

(一)、简单介绍

该服务接口可实现rtsp协议转换为rtmp协议且可以实现rtmp直播流发布到nginx流媒体服务器,其中最为重要的是如何实现通过参数生成ffmpeg命令并执行,且可以通过接口进行控制ffmpeg命令的停止

(二)、实现ffmpeg接口化服务架构设计

push端接口化管理

一、接口化调用

1、采用多线程方式,每次调用push端口开启一个主进程及两个输出线程 2、可以对每个push端(线程)进行开启和关闭的控制 3、统一接口参数,对ffmpeg的命令做到参数可控制

二、架构设计

1、服务接口

PushManager提供push(开启一个push处理器),closePush(关闭push处理器),viewAppName接口(查看当前已经开启的应用)

1.1、应用名和push处理器的关系

一个处理器对应一个应用名

1.2、push处理器

一个处理器对应一个push主进程和两个输出线程

2、主进程控制

2.1、主进程开启

服务接口调用push处理器开启push主进程,主进程会自动开启两个输出线程用于消息输出, 开启后会将主进程Process和两个输出线程OutHandler通过map返回给服务接口。

2.2、主进程关闭

主进程可通过Process的destroy方法进行安全关闭

3、输出线程控制

3.1、输出线程开启

输出线程从主进程获取到输出流进行输出

3.2、输出线程关闭

输出线程重写了destory方法,用于安全的关闭输出线程

4、持久层控制

持久层分为两个:

1、appName(应用名)-pushId(push处理器的ID)对应关系

用于维护应用名和push处理器ID的对应关系,pushId为随机生成id

2、pushId-主进程-输出线程对应关系

主要用于存放主进程(Process)和两个输出线程,建立两者对应关系,方便服务接口管理

(三)、代码实现

1、PushManager实现

/**
 * 实现push管理器的push,delete,view服务
 * 
 * @author eguid
 * @see PushMangerImpl
 * @since jdk1.7
 */

public class PushManagerImpl implements PushManager
{
    /**
     * 引用push处理器
     */
    private PushHandler pusher = new PushHandlerImpl();
    /**
     * 管理应用名和push处理器之间的关系
     */
    private PushId_AppRelshipDao pard=new PushId_AppRelshipDaoImpl();
    /**
     * 管理处理器的主进程Process及两个输出线程的关系
     */
    private HandlerDao hd = new HandlerDaoImpl();
    
    public void setPusher(PushHandler pusher)
    {
        this.pusher = pusher;
    }

    public void setPard(PushId_AppRelshipDao pard)
    {
        this.pard = pard;
    }

    public void setHd(HandlerDao hd)
    {
        this.hd = hd;
    }

    @Override
    public String push(Map<String, Object> map)
    {
        if(map==null||map.isEmpty()||!map.containsKey("appName"))
        {
            return null;
        }
        String appName=null;
        ConcurrentMap<String, Object> resultMap = null;
        try
        {
            appName=(String)map.get("appName");
            if(appName!=null&&"".equals(appName.trim()))
            {
                return null;
            }
            resultMap = pusher.push(map);
            // 生成一个标识该命令行线程集的key
            String pushId = UUID.randomUUID().toString();
            hd.set(pushId, resultMap);
            pard.set(appName, pushId);
        }
        catch (IOException e)
        {
            // 暂时先写这样,后期加日志
            System.err.println("发生一个异常" + e.getMessage());
        }
        return appName;
    }

    @Override
    public void closePush(String appName)
    {
        String pushId=null;
        if(pard.isHave(appName))
        {
            pushId= pard.getPushId(appName);
        }
        if (pushId!=null&&hd.isHave(pushId))
        {
            ConcurrentMap<String, Object> map = hd.get(pushId);
            //关闭两个线程
            ((OutHandler)map.get("error")).destroy();
            ((OutHandler)map.get("info")).destroy();
            //暂时先这样写,后期加日志
            System.out.println("停止命令-----end commond");
            //关闭命令主进程
            ((Process)map.get("process")).destroy();
            //删除处理器与线程对应关系表
            hd.delete(pushId);
            //删除应用名对应关系表
            pard.delete(appName);
        }
    }

    @Override
    public List<String> viewAppName()
    {
        return pard.getAll();
    }

2、pushHandler实现(push处理器)

/**
 * 提供解析参数生成ffmpeg命令并处理push操作
 * @see PushHandlerImpl
 * @since jdk1.7
 */

public class PushHandlerImpl implements PushHandler
{
    /*
     * "ffmpeg -i "+ "rtsp://admin:admin@192.168.2.236:37779/cam/realmonitor?channel=1&subtype=0 "+" -f flv -r 25 -s 640x360 -an" + " rtmp://192.168.30.21/live/test"
     * 推送流格式: name:应用名;input:接收地址;output:推送地址;fmt:视频格式;fps:视频帧率;rs:视频分辨率;disableAudio:是否开启音频
     */
    @Override
    public ConcurrentMap<String, Object> push(Map<String, Object> paramMap)
        throws IOException
    {
        // 从map里面取数据,组装成命令
        String comm = getComm4Map(paramMap);
        ConcurrentMap<String, Object> resultMap = null;
        // 执行命令行

        final Process proc = Runtime.getRuntime().exec(comm);
        System.out.println("执行命令----start commond");
        OutHandler errorGobbler = new OutHandler(proc.getErrorStream(), "Error");
        OutHandler outputGobbler = new OutHandler(proc.getInputStream(), "Info");

        errorGobbler.start();
        outputGobbler.start();
        // 返回参数
        resultMap = new ConcurrentHashMap<String, Object>();
        resultMap.put("info", outputGobbler);
        resultMap.put("error", errorGobbler);
        resultMap.put("process", proc);
        return resultMap;
    }

    /**
     * 通过解析参数生成可执行的命令行字符串;
     * name:应用名;input:接收地址;output:推送地址;fmt:视频格式;fps:视频帧率;rs:视频分辨率;disableAudio:是否开启音频
     * 
     * @param paramMap
     * @return 命令行字符串
     */
    protected String getComm4Map(Map<String, Object> paramMap)
    {
        // -i:输入流地址或者文件绝对地址
        StringBuilder comm = new StringBuilder("ffmpeg -i ");
        // 是否有必输项:输入地址,输出地址,应用名
        if (paramMap.containsKey("input") && paramMap.containsKey("output")
            && paramMap.containsKey("appName"))
        {
            comm.append(paramMap.get("input")).append(" ");
            // -f :转换格式,默认flv
            comm.append(" -f ").append(paramMap.containsKey("fmt") ? paramMap.get("fmt") : "flv").append(" ");
            // -r :帧率,默认25
            comm.append("-r ").append(paramMap.containsKey("fps") ? paramMap.get("fps") : "30").append(" ");
            // -s 分辨率 默认是原分辨率
            comm.append("-s ").append(paramMap.containsKey("rs") ? paramMap.get("rs") : "").append(" ");
            // -an 禁用音频 
            comm.append("-an ").append(paramMap.containsKey("disableAudio") && ((Boolean)paramMap.get("disableAudio")) ? "-an" : "").append(" ");
            // 输出地址
            comm.append(paramMap.get("output"));
            //发布的应用名
            comm.append(paramMap.get("appName"));
            //一个视频源,可以有多个输出,第二个输出为拷贝源视频输出,不改变视频的各项参数并且命名为应用名+HD
            comm.append(" ").append(" -vcodec copy -f flv -an ").append(paramMap.get("output")).append(paramMap.get("appName")).append("HD");
            System.out.println(comm.toString());
            return comm.toString();
        }
        else
        {
            throw new RuntimeException("输入流地址不能为空!");
        }

    }
}

3、OutHandler(输出线程)

**
 * 用于输出命令行主进程的消息线程(必须开启,否则命令行主进程无法正常执行) 重要:该类重写了destroy方法,用于安全的关闭该线程
 * 
 * @author eguid
 * @see OutHandler
 * @since jdk1.7
 */

public class OutHandler extends Thread
{
    // 控制状态
    volatile boolean status = true;

    BufferedReader br = null;

    String type = null;

    public OutHandler(InputStream is, String type)
    {
        br = new BufferedReader(new InputStreamReader(is));
        this.type = type;
    }

    /**
     * 重写线程销毁方法,安全的关闭线程
     */
    @Override
    public void destroy()
    {
        status = false;
    }

    /**
     * 执行输出线程
     */
    @Override
    public void run()
    {
        String msg = null;
        try
        {
            while (status)
            {

                if ((msg = br.readLine()) != null)
                {
                    System.out.println(type + "消息:" + msg);
                }
            }
        }
        catch (IOException e)
        {
            e.printStackTrace();
        }
    }

}

4、两个dao层接口,方便后期实现该接口并实现持久化

4.1、主进程(Process)和两个输出线程Dao

/**
 * 命令行执行处理器缓存,方便管理处理器的开启和关闭
 * @author eguid  
 * @see HandlerDao  
 * @since  jdk1.7
 */

public interface HandlerDao
{
    /**
     *  获取某个处理器
     * @param pushId
     * @return
     */
    public ConcurrentMap get(String pushId);
    /**
     * 存放一个处理器
     * @param handlerMap
     */
    public void set(String key, ConcurrentMap<String, Object> resultMap);
    /**
     * 获取全部处理器的id
     * @return
     */
    public ConcurrentMap getAll();
    /**
     * 删除某个处理器
     * @param pushId
     */
    public void delete(String pushId);
    /**
     * 是否存在key
     */
    public boolean isHave(String pushId);
}

4.2、应用名-pushId对应关系Dao

/**
 * 用于维护管理应用名与pushId的关系对应
 * @author eguid 
 * @see PushId_AppRelshipDao  
 * @since  jdk1.7
 */
public interface PushId_AppRelshipDao
{
/**
 *  获取应用名对应的pushId
 * @param appName
 * @return pushId
 */
public String getPushId(String appName);
/**
 *  插入一个应用名和pushId对应
 * @param appName
 * @param pushId
 */
public void set(String appName,String pushId);
/**
 * 通过应用名删除对应关系
 * @param appName
 */
public void delete(String appName);
/**
 * 获取全部应用
 */
public List<String> getAll();
/**
 * 是否存在应用名
 * @param appName
 * @return true:存在;false:不存在
 */
public boolean isHave(String appName);
}

下一篇将介绍一些支持rtmp直播的播放器

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 惊呆了,Spring Boot居然这么耗内存!

    Spring Boot总体来说,搭建还是比较容易的,特别是Spring Cloud全家桶,简称亲民微服务,但在发展趋势中,容器化技术已经成熟,面对巨耗内存的Sp...

    芋道源码
  • STL 源码剖析之动态数组 vector

    vector 的数据安排以及操作方式,与 array 非常相似。两者的唯一差别在于空间的运用的灵活性,array 是静态的,一旦配置了就不能改变,而 vecto...

    公众号guangcity
  • 工作了几年还不知道Java虚拟机?(图文详解JVM)

    Java编程指南
  • Skywalking之Java Agent基础

    之前的文章提到,要用 Skywalking 监控一个应用,需要在其 VM 参数中添加 “-javaagent:skywalking-agent.jar”(省略s...

    黄泽杰
  • 纳尼,Java 存在内存泄泄泄泄泄泄漏吗?

    Java 最牛逼的一个特性就是垃圾回收机制,不用像 C++ 需要手动管理内存,所以作为 Java 程序员很幸福,只管 New New New 即可,反正 Jav...

    奋斗蒙
  • 聊聊整体性学习方法

    「整体性学习方法」是在一本叫做《如何高效学习》的书中看到的。这本书的作者是个老外,他用一年就学完了四年的麻省理工课程。而这本书正是其这一年来的学习心得,书中介绍...

    范蠡
  • Elasticsearch 7.4.0 发布,分布式搜索和数据分析引擎

    Elasticsearch 7.4.0 发布了,Elasticsearch 是一个分布式的 RESTful 风格的搜索和数据分析引擎。

    芋道源码
  • Spring Boot(十八):使用 Spring Boot 集成 FastDFS

    上篇文章介绍了如何使用 Spring Boot 上传文件,这篇文章我们介绍如何使用 Spring Boot 将文件上传到分布式文件系统 FastDFS 中。

    纯洁的微笑
  • C++ STL源码剖析 tr1与std array

    C++ tr1全称Technical Report 1,是针对C++标准库的第一次扩展。即将到来的下一个版本的C++标准c++0x会包括它,以及一些语言本身的扩...

    公众号guangcity
  • iOS内存管理-基本概念整理

    主要内容: 1.内存区域划分 2.内存管理/引用计数 3.MRC手动管理引用计数 4.ARC自动引用计数 5.内存泄漏问题 6.野指针问题

    梧雨北辰

扫码关注云+社区

领取腾讯云代金券