(aiohttp-asyncio-FFmpeg-Docker-SRS)实现异步摄像头转码服务器

本文主要是介绍(aiohttp-asyncio-FFmpeg-Docker-SRS)实现异步摄像头转码服务器,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

1. 背景介绍

在先前的博客文章中,我们已经搭建了一个基于SRS的流媒体服务器。现在,我们希望通过Web接口来控制这个服务器的行为,特别是对于正在进行的 RTSP 转码任务的管理。这将使我们能够在不停止整个服务器的情况下,动态地启动或停止摄像头的转码过程。

Docker部署 SRS rtmp/flv流媒体服务器-CSDN博客文章浏览阅读360次,点赞7次,收藏5次。SRS(Simple Realtime Server)是一款开源的流媒体服务器,具有高性能、高可靠性、高灵活性的特点,能够支持直播、点播、转码等多种流媒体应用场景。SRS 不仅提供了流媒体服务器,还提供了适用于多种平台的客户端 SDK 和在线转码等辅助服务,是一款十分强大的流媒体解决方案。https://blog.csdn.net/m0_56659620/article/details/135400510?spm=1001.2014.3001.5501

2. 技术选择

在选择技术方案时,考虑到构建视频流转码服务的需求,我们将采用Python编程语言,并结合asyncio和aiohttp库。这一选择基于异步框架的优势,以下是对异步框架和同步框架在视频流转码场景中的优缺点的明确总结:

异步框架的优势:

  • 高并发处理: 异步框架通过非阻塞方式处理请求,能够高效处理大量并发请求,确保系统在高负载下保持稳定性。
  • 异步I/O: 支持异步I/O操作,允许在等待I/O操作完成的同时继续处理其他请求,提高整体效率。
  • 资源利用率高: 能够更有效地利用系统资源,同时处理多个请求,提高视频转码效率。
  • 事件驱动: 采用事件驱动模型,适应实时性要求高的视频流处理,能够立即响应新的转码请求。

同步框架的缺点:

  • 阻塞: 阻塞调用可能导致整个程序停滞,尤其在处理大文件或网络请求时可能引发性能问题,特别是在高并发场景下。
  • 低并发: 每个请求需要独立的线程或进程,可能导致系统资源耗尽,降低并发处理能力,对于需要同时处理多个视频流的情况可能不够高效。

考虑到处理大量并发请求、提高系统性能和响应性的需求,采用异步框架是更为合适的选择。异步框架的高并发处理能力、异步I/O支持、高资源利用率以及事件驱动的特性使其更适用于实时性要求较高的视频流转码服务。

3. 代码实现(必须在linux系统运行,4步骤为部署攻略)

3.1 导入必要的库

首先,我们导入所需的库,包括asyncio、aiohttp、aiohttp_cors和logging。

import asyncio
from aiohttp import web
import aiohttp_cors
import logging

3.2 设置日志

logging.basicConfig(level=logging.INFO)

3.3 配置并发控制和任务跟踪

设置最大同时运行的ffmpeg子进程数量,并使用Semaphore限制并发进程数量。同时,使用字典跟踪正在进行的转码任务。

MAX_CONCURRENT_PROCESSES = 5
semaphore = asyncio.Semaphore(MAX_CONCURRENT_PROCESSES)
transcoding_tasks = {}

3.4 定义启动和停止转码任务的方法

定义启动和停止 RTSP 转码任务的方法

# 开始转码方法
async def perform_transcoding(ip, camera_id, rtmp_server):# 检查相同RTSP是否已有子进程在处理if camera_id in transcoding_tasks:return transcoding_tasks[camera_id]# 使用Semaphore限制并发进程数量async with semaphore:# 实际的转码操作,这里需要调用ffmpeg或其他工具ffmpeg_command = ['ffmpeg','-rtsp_transport', 'tcp','-i', ip,'-c:v', 'libx264','-c:a', 'aac','-f', 'flv',f'{rtmp_server}/live/livestream{camera_id}']# 创建异步子进程process = await asyncio.create_subprocess_exec(*ffmpeg_command)# 将任务添加到字典中transcoding_tasks[camera_id] = process# 等待子进程完成await process.communicate()# 从字典中移除已完成的任务transcoding_tasks.pop(camera_id, None)# 停止转码方法
async def stop_transcoding(camera_id):# 停止转码任务if camera_id in transcoding_tasks:process = transcoding_tasks[camera_id]process.terminate()  # 发送终止信号await process.wait()  # 等待进程结束# 从字典中移除已停止的任务transcoding_tasks.pop(camera_id, None)

3.5 定义Web接口路由

定义Web接口路由,包括启动摄像头转码、停止摄像头转码和停止所有摄像头转码的路由。

# 开始转码任务
async def play_camera(request):data = await request.post()# 从表单数据中获取摄像头的ID和rtsp流camera_id = data.get('id')rtsp = data.get('ip')# 这里设置你的 RTMP 服务器地址rtmp_server = 'rtmp://192.168.14.93:1935'# 执行实际的转码操作task = await perform_transcoding(rtsp, camera_id, rtmp_server)# 返回包含转码后的RTMP URL的JSON响应rtmp_url = f'http://192.168.14.93:8080/live/livestream{camera_id}.flv'return web.json_response({'message': '转码启动成功', 'flv_data': rtmp_url})# 停止转码任务
async def stop_camera(request):data = await request.post()camera_id = data.get('id')# 停止指定摄像头的转码任务await stop_transcoding(camera_id)return web.json_response({'code':200,'message': '转码停止成功'})# 如果页面进行刷新或者关闭停止全部转码任务
async def stop_all_camera(request):# 获取所有正在运行的任务的列表tasks = [stop_transcoding(camera_id) for camera_id in transcoding_tasks.keys()]# 并发停止所有任务await asyncio.gather(*tasks)# 清空字典,表示所有任务都已停止transcoding_tasks.clear()return web.json_response({'code':200,'message': '转码停止成功'})

3.6 创建Web应用和配置CORS

创建Web应用,配置CORS(跨域资源共享)中间件,以确保接口可以被跨域访问。

app = web.Application()# CORS配置
cors = aiohttp_cors.setup(app, defaults={"*": aiohttp_cors.ResourceOptions(allow_credentials=True,expose_headers="*",allow_headers="*",)
})

3.7 添加Web接口路由

添加Web接口路由,包括启动摄像头转码、停止摄像头转码和停止所有摄像头转码的路由。

app.router.add_route('POST', '/play_camera', play_camera)  # 开始转码任务路由
app.router.add_route('POST', '/stop_camera', stop_camera)  # 停止转码任务路由
app.router.add_route('POST', '/stop_all_camera', stop_all_camera)  # 停止全部转码任务路由

3.8 添加CORS中间件

添加CORS中间件,确保接口可以被跨域访问。

# 添加 CORS 中间件
for route in list(app.router.routes()):cors.add(route)

3.9 运行Web应用

运行Web应用,监听指定的主机和端口。

if __name__ == '__main__':web.run_app(app, host='0.0.0.0', port=7000,access_log=logging.getLogger())

 3.10 完整代码

import asyncio
from aiohttp import web
import aiohttp_cors
import logging# 设置日志级别
logging.basicConfig(level=logging.INFO)# 最大同时运行的ffmpeg子进程数量
MAX_CONCURRENT_PROCESSES = 5# 使用Semaphore限制并发进程数量
semaphore = asyncio.Semaphore(MAX_CONCURRENT_PROCESSES)# 字典用于跟踪正在进行的转码任务
transcoding_tasks = {}# 开始转码方法
async def perform_transcoding(ip, camera_id, rtmp_server):# 检查相同RTSP是否已有子进程在处理if camera_id in transcoding_tasks:return transcoding_tasks[camera_id]# 使用Semaphore限制并发进程数量async with semaphore:# 实际的转码操作,这里需要调用ffmpeg或其他工具ffmpeg_command = ['ffmpeg','-rtsp_transport', 'tcp','-i', ip,'-c:v', 'libx264','-c:a', 'aac','-f', 'flv',f'{rtmp_server}/live/livestream{camera_id}']# 创建异步子进程process = await asyncio.create_subprocess_exec(*ffmpeg_command)# 将任务添加到字典中transcoding_tasks[camera_id] = process# 等待子进程完成await process.communicate()# 从字典中移除已完成的任务transcoding_tasks.pop(camera_id, None)# 停止转码方法
async def stop_transcoding(camera_id):# 停止转码任务if camera_id in transcoding_tasks:process = transcoding_tasks[camera_id]process.terminate()  # 发送终止信号await process.wait()  # 等待进程结束# 从字典中移除已停止的任务transcoding_tasks.pop(camera_id, None)# 开始转码任务
async def play_camera(request):data = await request.post()# 从表单数据中获取摄像头的ID和rtsp流camera_id = data.get('id')rtsp = data.get('ip')# 这里设置你的 RTMP 服务器地址rtmp_server = 'rtmp://192.168.14.93:1935'# 执行实际的转码操作task = await perform_transcoding(rtsp, camera_id, rtmp_server)# 返回包含转码后的RTMP URL的JSON响应rtmp_url = f'http://192.168.14.93:8080/live/livestream{camera_id}.flv'return web.json_response({'message': '转码启动成功', 'flv_data': rtmp_url})# 停止转码任务
async def stop_camera(request):data = await request.post()camera_id = data.get('id')# 停止指定摄像头的转码任务await stop_transcoding(camera_id)return web.json_response({'code':200,'message': '转码停止成功'})# 如果页面进行刷新或者关闭停止全部转码任务
async def stop_all_camera(request):# 获取所有正在运行的任务的列表tasks = [stop_transcoding(camera_id) for camera_id in transcoding_tasks.keys()]# 并发停止所有任务await asyncio.gather(*tasks)# 清空字典,表示所有任务都已停止transcoding_tasks.clear()return web.json_response({'code':200,'message': '转码停止成功'})app = web.Application()# CORS配置
cors = aiohttp_cors.setup(app, defaults={"*": aiohttp_cors.ResourceOptions(allow_credentials=True,expose_headers="*",allow_headers="*",)
})app.router.add_route('POST', '/play_camera', play_camera)  # 开始转码任务路由
app.router.add_route('POST', '/stop_camera', stop_camera)  # 停止转码任务路由
app.router.add_route('POST', '/stop_all_camera', stop_all_camera)  # 停止全部转码任务路由# 添加 CORS 中间件
for route in list(app.router.routes()):cors.add(route)if __name__ == '__main__':web.run_app(app, host='0.0.0.0', port=7000,access_log=logging.getLogger())

4. 部署(Docker环境)

部署所需Dockerfile文件代码如下

FROM python:3.7-slimWORKDIR /appCOPY requirements.txt .RUN apt-get update \&& apt-get install -y ffmpeg \&& rm -rf /var/lib/apt/lists/* \&& pip install --no-cache-dir -r requirements.txtCOPY . .CMD ["python", "async_io_io.py"]

部署所需requirements.txt如下

aiohttp
aiohttp-cors
ffmpeg

根目录进行打包及启动

请求接口实现转码

5. 总结

通过以上的步骤,我们成功构建了一个流媒体服务器控制接口,可以通过Web接口实现对摄像头的 RTSP 转码任务的动态管理。这个接口可以集成到现有的流媒体服务器中,提供更多控制和管理的可能性。

这篇关于(aiohttp-asyncio-FFmpeg-Docker-SRS)实现异步摄像头转码服务器的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/584414

相关文章

Java实现时间与字符串互相转换详解

《Java实现时间与字符串互相转换详解》这篇文章主要为大家详细介绍了Java中实现时间与字符串互相转换的相关方法,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 目录一、日期格式化为字符串(一)使用预定义格式(二)自定义格式二、字符串解析为日期(一)解析ISO格式字符串(二)解析自定义

opencv图像处理之指纹验证的实现

《opencv图像处理之指纹验证的实现》本文主要介绍了opencv图像处理之指纹验证的实现,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学... 目录一、简介二、具体案例实现1. 图像显示函数2. 指纹验证函数3. 主函数4、运行结果三、总结一、

Springboot处理跨域的实现方式(附Demo)

《Springboot处理跨域的实现方式(附Demo)》:本文主要介绍Springboot处理跨域的实现方式(附Demo),具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不... 目录Springboot处理跨域的方式1. 基本知识2. @CrossOrigin3. 全局跨域设置4.

Spring Boot 3.4.3 基于 Spring WebFlux 实现 SSE 功能(代码示例)

《SpringBoot3.4.3基于SpringWebFlux实现SSE功能(代码示例)》SpringBoot3.4.3结合SpringWebFlux实现SSE功能,为实时数据推送提供... 目录1. SSE 简介1.1 什么是 SSE?1.2 SSE 的优点1.3 适用场景2. Spring WebFlu

基于SpringBoot实现文件秒传功能

《基于SpringBoot实现文件秒传功能》在开发Web应用时,文件上传是一个常见需求,然而,当用户需要上传大文件或相同文件多次时,会造成带宽浪费和服务器存储冗余,此时可以使用文件秒传技术通过识别重复... 目录前言文件秒传原理代码实现1. 创建项目基础结构2. 创建上传存储代码3. 创建Result类4.

SpringBoot日志配置SLF4J和Logback的方法实现

《SpringBoot日志配置SLF4J和Logback的方法实现》日志记录是不可或缺的一部分,本文主要介绍了SpringBoot日志配置SLF4J和Logback的方法实现,文中通过示例代码介绍的非... 目录一、前言二、案例一:初识日志三、案例二:使用Lombok输出日志四、案例三:配置Logback一

Python如何使用__slots__实现节省内存和性能优化

《Python如何使用__slots__实现节省内存和性能优化》你有想过,一个小小的__slots__能让你的Python类内存消耗直接减半吗,没错,今天咱们要聊的就是这个让人眼前一亮的技巧,感兴趣的... 目录背景:内存吃得满满的类__slots__:你的内存管理小助手举个大概的例子:看看效果如何?1.

Python+PyQt5实现多屏幕协同播放功能

《Python+PyQt5实现多屏幕协同播放功能》在现代会议展示、数字广告、展览展示等场景中,多屏幕协同播放已成为刚需,下面我们就来看看如何利用Python和PyQt5开发一套功能强大的跨屏播控系统吧... 目录一、项目概述:突破传统播放限制二、核心技术解析2.1 多屏管理机制2.2 播放引擎设计2.3 专

Python实现无痛修改第三方库源码的方法详解

《Python实现无痛修改第三方库源码的方法详解》很多时候,我们下载的第三方库是不会有需求不满足的情况,但也有极少的情况,第三方库没有兼顾到需求,本文将介绍几个修改源码的操作,大家可以根据需求进行选择... 目录需求不符合模拟示例 1. 修改源文件2. 继承修改3. 猴子补丁4. 追踪局部变量需求不符合很

idea中创建新类时自动添加注释的实现

《idea中创建新类时自动添加注释的实现》在每次使用idea创建一个新类时,过了一段时间发现看不懂这个类是用来干嘛的,为了解决这个问题,我们可以设置在创建一个新类时自动添加注释,帮助我们理解这个类的用... 目录前言:详细操作:步骤一:点击上方的 文件(File),点击&nbmyHIgsp;设置(Setti