您当前的位置： 首页 > 慢生活 > 程序人生 网站首页程序人生

在php中如何实现rtsp流转换成mp4文件并且异步多进程且不重复处理任务？

发布时间：2025-02-23 22:06:13编辑：雪饮阅读（）

前番研究过将rtsp流转推为rtsp流，并且实现了基于nginx的多推拉流地址，那么实际情况下，这种方案需要服务器配置比较高，我测试发现我的2核4G的centos8勉强抗住2推流2拉流。那么有一些需求，例如我这次面临的需求是客户要求在小程序中播放，这种方案不是很好，那么好在是客户是可以接受下载视频到本地的。那么这次就是说要实现将rtsp流转存本地，这个其实前番也有讲过，只是这次是面对php中的处理难题。
 
在php中一般大都是基于nginx的web环境居多，这种环境下或许处理很短小的视频问题不大，对于长达可能是2小时的视频来说，基本算是硬伤了。最好的解决办法就是在命令行里面运行php进行调用转码。
 
在命令行里面运行转码，那么就要不断的去查询数据库中是否有转码任务，比如每间隔1分钟就去查询一下，然后取出来进行转码。但是这样来说你还要处理php与数据库的长连接，以及连接中断重试等多种方案。
再一个就是多进程问题，如果你的服务器能够同时处理多个转码任务，那么多开一个进程自然是好的，但是多开一个进程进行处理，那么如果在某一个时刻两个进程查询到同一个数据记录处理了同一个任务，这也造成了重复处理任务的问题。当然可能有数据库的锁机制之类解决方案的，但是我这里先不提这个方案。
 
基于以上这些问题：
首先来说每间隔时间来取任务的不优雅的解决方案是使用原子性队列，一次性只能取一个任务在同一时刻。
这第二个就是进程多开的情况，也就是同时进行多个转码任务，当你运行一个进程时候，你推多个任务到队列，那么进程必须等上一个任务结束才处理下一个，那么这个任务其实你用php去循环取队列的任务并将该php脚本多进程运行即可，但需要注意的是我这里实践了基于fastadmin，也就是好像thinkphp5.0(5.0.27),怎么查看thinkphp版本呢？
在fastadmin目录中thinkphp目录下的base.php中就能查看到thinkphp的版本。
就是我基于该thinkphp的命令行例如我执行php think “你的thinkphp命令名”这种方式如果运行多个，仍旧是会出现多个重复消费的情况，具体缘由不是很清楚。
具体thinkphp5的命令行使用可以参考
https://doc.thinkphp.cn/v5_0/zidingyiminglingxing.html
那么所以我的方案还是原生php方案，至于数据库操作或者redis操作这种涉及到连接的我都是临时用临时建立连接，那么数据库肯定用的最多，所以命令行里面不用框架的情况下，原生php也是不太好写的，注意是太啰嗦（毕竟原生sql太啰嗦），所以我采取的是用guzzlehttp客户端来请求我们实际web项目中我们自定义的api来完成。
 
那么最后你可以手动运行这个命令多次，当然这种长期处理的任务的进程肯定是阻塞的，所以你可以后台运行他或者开多个会话去运行，或者用Supervisor来守护进程，如在宝塔上安装Supervisor然后以你的命令来创建进程的时候可以直接填写进程数量。
不过我最终还是单进程运行，我们服务器虽然4核8g，但是要考虑我们服务器还有其它业务系统的运行，开2个进程都有点系统运行堵塞的情况。
接下来我就分享下我这个代码吧：
<?php
define('DS', DIRECTORY_SEPARATOR);
define('ROOT_PATH', __DIR__.DS);
define('LOG_PATH', __DIR__."/runtime/log/");
define('IS_CLI', PHP_SAPI == 'cli' ? true : false);
defined('ENV_PREFIX') or define('ENV_PREFIX', 'PHP_');
use think\Log;
use GuzzleHttp\Client;
require_once __DIR__ . '/vendor/autoload.php';
$configArr=require_once __DIR__ .'/application/database.php';
$redis = new \Redis();
$redis->connect('127.0.0.1', 6379);
while (true)
{
    try{
        //从队列提取数据,超时时间5秒
        $content = $redis->brPop('ffmpeg_native', 5);
        if ($content)
        {
 
            $queue=$content[0];
 
            $data = json_decode($content['1'], true);
 
            $user=$configArr["user"];
            $password=$configArr["password"];
            $hostname=$configArr["hostname"];
            $hostport=$configArr["hostport"];
            $database=$configArr["database"];
 
 
 
 
            $info="开始处理任务:".$data["jobId"].",命令:".$data["command"].PHP_EOL;
            Log::record($info, 'info');
            echo $info;
            exec($data["command"], $output, $return_var);
            $apiUrl=$data["REQUEST_SCHEME"].'://'.$data["HTTP_HOST"].'/api/haiv2/jobUpdateToFinishStatus';
            if ($return_var === 0) {
                $info="任务.".$data["jobId"].",命令执行完成，去请求url".$apiUrl."更新任务完成标记".PHP_EOL;
                echo $info;
                Log::record($info, 'info');
                $client = new Client();
                $response = $client->request('POST', $apiUrl, [
                    'form_params' => [
                        'finish_url' => $data["finish_url"],
                        'jobId' => $data["jobId"],
                        'type' => 1
                    ]
                ]);
                $body = $response->getBody();
                $bodyStr = (string)$body;
                $info="任务.".$data["jobId"].",命令执行完成，更新任务完成标记响应数据:".$bodyStr.PHP_EOL;
                echo $info;
                Log::record($info, 'info');
            }
            else {
                $response = $client->request('POST', $apiUrl, [
                    'form_params' => [
                        'jobId' => $data["jobId"],
                        'type' => 0
                    ]
                ]);
                $bodyStr = (string)$body;
                $info="任务.".$data["jobId"].",命令执行失败，更新任务完成标记响应数据:".$bodyStr.PHP_EOL;
                echo $info;
                Log::record($info, 'info');
 
                $info="任务.".$data["jobId"].",命令执行失败，错误码为:".$return_var.PHP_EOL;
                echo $info;
                Log::record($info, 'info');
 
                $info="任务.".$data["jobId"].",命令执行失败，exec=>output:".var_export($output,true).PHP_EOL;
                echo $info;
                Log::record($info, 'info');
                //重试
                $apiUrl=$data["REQUEST_SCHEME"].'://'.$data["HTTP_HOST"].'/api/haiv2/taskRetry?id='.$data["jobId"];
                file_get_contents($apiUrl);
 
            }
 
 
        }
    }
    catch(\Exception $e){
        $info="更新命令处理结束标记时出错了：file:".$e->getFile().",line:".$e->getLine().",message:".$e->getMessage().PHP_EOL;
        echo $info;
        Log::record($info, 'info');
    }
 
}
那么接下来就是我向任务队列中推送任务的关键代码如：
$queueData["jobId"]=$jobId;
        $queueData["command"]=$command;
        $queueData["REQUEST_SCHEME"]=$_SERVER['REQUEST_SCHEME'];
        $queueData["HTTP_HOST"]=$_SERVER['HTTP_HOST'];
        $queueData["finish_url"]=$_SERVER['REQUEST_SCHEME'].":".$_SERVER["HTTP_HOST"]."/orderVideoTmp/".$fileId.".mp4";
 
        $redis = new Redis;
        $redis->connect('127.0.0.1', 6379);
        $res = $redis->lPush('ffmpeg_native', json_encode($queueData));
 $redis->close();
        if($res){
            $this->success('下载任务已发送，预计完成时间:'.$elapsedTime.",请稍后查看");
        }