1. 背景介紹
在先前的博客文章中，我們已經(jīng)搭建了一個(gè)基于SRS的流媒體服務(wù)器?，F(xiàn)在，我們希望通過Web接口來控制這個(gè)服務(wù)器的行為，特別是對(duì)于正在進(jìn)行的 RTSP 轉(zhuǎn)碼任務(wù)的管理。這將使我們能夠在不停止整個(gè)服務(wù)器的情況下，動(dòng)態(tài)地啟動(dòng)或停止攝像頭的轉(zhuǎn)碼過程。
Docker部署 SRS rtmp/flv流媒體服務(wù)器-CSDN博客文章瀏覽閱讀360次，點(diǎn)贊7次，收藏5次。SRS（Simple Realtime Server）是一款開源的流媒體服務(wù)器，具有高性能、高可靠性、高靈活性的特點(diǎn)，能夠支持直播、點(diǎn)播、轉(zhuǎn)碼等多種流媒體應(yīng)用場(chǎng)景。SRS 不僅提供了流媒體服務(wù)器，還提供了適用于多種平臺(tái)的客戶端 SDK 和在線轉(zhuǎn)碼等輔助服務(wù)，是一款十分強(qiáng)大的流媒體解決方案。https://blog.csdn.net/m0_56659620/article/details/135400510?spm=1001.2014.3001.5501
2. 技術(shù)選擇
在選擇技術(shù)方案時(shí)，考慮到構(gòu)建視頻流轉(zhuǎn)碼服務(wù)的需求，我們將采用Python編程語言，并結(jié)合asyncio和aiohttp庫。這一選擇基于異步框架的優(yōu)勢(shì)，以下是對(duì)異步框架和同步框架在視頻流轉(zhuǎn)碼場(chǎng)景中的優(yōu)缺點(diǎn)的明確總結(jié)：
異步框架的優(yōu)勢(shì)：
- 高并發(fā)處理： 異步框架通過非阻塞方式處理請(qǐng)求，能夠高效處理大量并發(fā)請(qǐng)求，確保系統(tǒng)在高負(fù)載下保持穩(wěn)定性。
- 異步I/O： 支持異步I/O操作，允許在等待I/O操作完成的同時(shí)繼續(xù)處理其他請(qǐng)求，提高整體效率。
- 資源利用率高： 能夠更有效地利用系統(tǒng)資源，同時(shí)處理多個(gè)請(qǐng)求，提高視頻轉(zhuǎn)碼效率。
- 事件驅(qū)動(dòng)： 采用事件驅(qū)動(dòng)模型，適應(yīng)實(shí)時(shí)性要求高的視頻流處理，能夠立即響應(yīng)新的轉(zhuǎn)碼請(qǐng)求。
同步框架的缺點(diǎn)：
- 阻塞： 阻塞調(diào)用可能導(dǎo)致整個(gè)程序停滯，尤其在處理大文件或網(wǎng)絡(luò)請(qǐng)求時(shí)可能引發(fā)性能問題，特別是在高并發(fā)場(chǎng)景下。
- 低并發(fā)： 每個(gè)請(qǐng)求需要獨(dú)立的線程或進(jìn)程，可能導(dǎo)致系統(tǒng)資源耗盡，降低并發(fā)處理能力，對(duì)于需要同時(shí)處理多個(gè)視頻流的情況可能不夠高效。
考慮到處理大量并發(fā)請(qǐng)求、提高系統(tǒng)性能和響應(yīng)性的需求，采用異步框架是更為合適的選擇。異步框架的高并發(fā)處理能力、異步I/O支持、高資源利用率以及事件驅(qū)動(dòng)的特性使其更適用于實(shí)時(shí)性要求較高的視頻流轉(zhuǎn)碼服務(wù)。
3. 代碼實(shí)現(xiàn)（必須在linux系統(tǒng)運(yùn)行，4步驟為部署攻略）
3.1 導(dǎo)入必要的庫
首先，我們導(dǎo)入所需的庫，包括asyncio、aiohttp、aiohttp_cors和logging。
import asyncio
from aiohttp import web
import aiohttp_cors
import logging
3.2 設(shè)置日志
logging.basicConfig(level=logging.INFO)3.3 配置并發(fā)控制和任務(wù)跟蹤
設(shè)置最大同時(shí)運(yùn)行的ffmpeg子進(jìn)程數(shù)量，并使用Semaphore限制并發(fā)進(jìn)程數(shù)量。同時(shí)，使用字典跟蹤正在進(jìn)行的轉(zhuǎn)碼任務(wù)。
MAX_CONCURRENT_PROCESSES = 5
semaphore = asyncio.Semaphore(MAX_CONCURRENT_PROCESSES)
transcoding_tasks = {}
3.4 定義啟動(dòng)和停止轉(zhuǎn)碼任務(wù)的方法
定義啟動(dòng)和停止 RTSP 轉(zhuǎn)碼任務(wù)的方法
# 開始轉(zhuǎn)碼方法
async def perform_transcoding(ip, camera_id, rtmp_server):
# 檢查相同RTSP是否已有子進(jìn)程在處理
if camera_id in transcoding_tasks:
return transcoding_tasks[camera_id]
# 使用Semaphore限制并發(fā)進(jìn)程數(shù)量
async with semaphore:
# 實(shí)際的轉(zhuǎn)碼操作，這里需要調(diào)用ffmpeg或其他工具
ffmpeg_command = [
'ffmpeg',
'-rtsp_transport', 'tcp',
'-i', ip,
'-c:v', 'libx264',
'-c:a', 'aac',
'-f', 'flv',
f'{rtmp_server}/live/livestream{camera_id}'
]
# 創(chuàng)建異步子進(jìn)程
process = await asyncio.create_subprocess_exec(*ffmpeg_command)
# 將任務(wù)添加到字典中
transcoding_tasks[camera_id] = process
# 等待子進(jìn)程完成
await process.communicate()
# 從字典中移除已完成的任務(wù)
transcoding_tasks.pop(camera_id, None)
# 停止轉(zhuǎn)碼方法
async def stop_transcoding(camera_id):
# 停止轉(zhuǎn)碼任務(wù)
if camera_id in transcoding_tasks:
process = transcoding_tasks[camera_id]
process.terminate() # 發(fā)送終止信號(hào)
await process.wait() # 等待進(jìn)程結(jié)束
# 從字典中移除已停止的任務(wù)
transcoding_tasks.pop(camera_id, None)3.5 定義Web接口路由
定義Web接口路由，包括啟動(dòng)攝像頭轉(zhuǎn)碼、停止攝像頭轉(zhuǎn)碼和停止所有攝像頭轉(zhuǎn)碼的路由。
# 開始轉(zhuǎn)碼任務(wù)
async def play_camera(request):
data = await request.post()
# 從表單數(shù)據(jù)中獲取攝像頭的ID和rtsp流
camera_id = data.get('id')
rtsp = data.get('ip')
# 這里設(shè)置你的 RTMP 服務(wù)器地址
rtmp_server = 'rtmp://192.168.14.93:1935'
# 執(zhí)行實(shí)際的轉(zhuǎn)碼操作
task = await perform_transcoding(rtsp, camera_id, rtmp_server)
# 返回包含轉(zhuǎn)碼后的RTMP URL的JSON響應(yīng)
rtmp_url = f'http://192.168.14.93:8080/live/livestream{camera_id}.flv'
return web.json_response({'message': '轉(zhuǎn)碼啟動(dòng)成功', 'flv_data': rtmp_url})
# 停止轉(zhuǎn)碼任務(wù)
async def stop_camera(request):
data = await request.post()
camera_id = data.get('id')
# 停止指定攝像頭的轉(zhuǎn)碼任務(wù)
await stop_transcoding(camera_id)
return web.json_response({'code':200,'message': '轉(zhuǎn)碼停止成功'})
# 如果頁面進(jìn)行刷新或者關(guān)閉停止全部轉(zhuǎn)碼任務(wù)
async def stop_all_camera(request):
# 獲取所有正在運(yùn)行的任務(wù)的列表
tasks = [stop_transcoding(camera_id) for camera_id in transcoding_tasks.keys()]
# 并發(fā)停止所有任務(wù)
await asyncio.gather(*tasks)
# 清空字典，表示所有任務(wù)都已停止
transcoding_tasks.clear()
return web.json_response({'code':200,'message': '轉(zhuǎn)碼停止成功'})3.6 創(chuàng)建Web應(yīng)用和配置CORS
創(chuàng)建Web應(yīng)用，配置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接口路由，包括啟動(dòng)攝像頭轉(zhuǎn)碼、停止攝像頭轉(zhuǎn)碼和停止所有攝像頭轉(zhuǎn)碼的路由。
app.router.add_route('POST', '/play_camera', play_camera) # 開始轉(zhuǎn)碼任務(wù)路由
app.router.add_route('POST', '/stop_camera', stop_camera) # 停止轉(zhuǎn)碼任務(wù)路由
app.router.add_route('POST', '/stop_all_camera', stop_all_camera) # 停止全部轉(zhuǎn)碼任務(wù)路由3.8 添加CORS中間件
添加CORS中間件，確保接口可以被跨域訪問。
# 添加 CORS 中間件
for route in list(app.router.routes()):
cors.add(route)3.9 運(yùn)行Web應(yīng)用
運(yùn)行Web應(yīng)用，監(jiān)聽指定的主機(jī)和端口。
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
# 設(shè)置日志級(jí)別
logging.basicConfig(level=logging.INFO)
# 最大同時(shí)運(yùn)行的ffmpeg子進(jìn)程數(shù)量
MAX_CONCURRENT_PROCESSES = 5
# 使用Semaphore限制并發(fā)進(jìn)程數(shù)量
semaphore = asyncio.Semaphore(MAX_CONCURRENT_PROCESSES)
# 字典用于跟蹤正在進(jìn)行的轉(zhuǎn)碼任務(wù)
transcoding_tasks = {}
# 開始轉(zhuǎn)碼方法
async def perform_transcoding(ip, camera_id, rtmp_server):
# 檢查相同RTSP是否已有子進(jìn)程在處理
if camera_id in transcoding_tasks:
return transcoding_tasks[camera_id]
# 使用Semaphore限制并發(fā)進(jìn)程數(shù)量
async with semaphore:
# 實(shí)際的轉(zhuǎn)碼操作，這里需要調(diào)用ffmpeg或其他工具
ffmpeg_command = [
'ffmpeg',
'-rtsp_transport', 'tcp',
'-i', ip,
'-c:v', 'libx264',
'-c:a', 'aac',
'-f', 'flv',
f'{rtmp_server}/live/livestream{camera_id}'
]
# 創(chuàng)建異步子進(jìn)程
process = await asyncio.create_subprocess_exec(*ffmpeg_command)
# 將任務(wù)添加到字典中
transcoding_tasks[camera_id] = process
# 等待子進(jìn)程完成
await process.communicate()
# 從字典中移除已完成的任務(wù)
transcoding_tasks.pop(camera_id, None)
# 停止轉(zhuǎn)碼方法
async def stop_transcoding(camera_id):
# 停止轉(zhuǎn)碼任務(wù)
if camera_id in transcoding_tasks:
process = transcoding_tasks[camera_id]
process.terminate() # 發(fā)送終止信號(hào)
await process.wait() # 等待進(jìn)程結(jié)束
# 從字典中移除已停止的任務(wù)
transcoding_tasks.pop(camera_id, None)
# 開始轉(zhuǎn)碼任務(wù)
async def play_camera(request):
data = await request.post()
# 從表單數(shù)據(jù)中獲取攝像頭的ID和rtsp流
camera_id = data.get('id')
rtsp = data.get('ip')
# 這里設(shè)置你的 RTMP 服務(wù)器地址
rtmp_server = 'rtmp://192.168.14.93:1935'
# 執(zhí)行實(shí)際的轉(zhuǎn)碼操作
task = await perform_transcoding(rtsp, camera_id, rtmp_server)
# 返回包含轉(zhuǎn)碼后的RTMP URL的JSON響應(yīng)
rtmp_url = f'http://192.168.14.93:8080/live/livestream{camera_id}.flv'
return web.json_response({'message': '轉(zhuǎn)碼啟動(dòng)成功', 'flv_data': rtmp_url})
# 停止轉(zhuǎn)碼任務(wù)
async def stop_camera(request):
data = await request.post()
camera_id = data.get('id')
# 停止指定攝像頭的轉(zhuǎn)碼任務(wù)
await stop_transcoding(camera_id)
return web.json_response({'code':200,'message': '轉(zhuǎn)碼停止成功'})
# 如果頁面進(jìn)行刷新或者關(guān)閉停止全部轉(zhuǎn)碼任務(wù)
async def stop_all_camera(request):
# 獲取所有正在運(yùn)行的任務(wù)的列表
tasks = [stop_transcoding(camera_id) for camera_id in transcoding_tasks.keys()]
# 并發(fā)停止所有任務(wù)
await asyncio.gather(*tasks)
# 清空字典，表示所有任務(wù)都已停止
transcoding_tasks.clear()
return web.json_response({'code':200,'message': '轉(zhuǎn)碼停止成功'})
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) # 開始轉(zhuǎn)碼任務(wù)路由
app.router.add_route('POST', '/stop_camera', stop_camera) # 停止轉(zhuǎn)碼任務(wù)路由
app.router.add_route('POST', '/stop_all_camera', stop_all_camera) # 停止全部轉(zhuǎn)碼任務(wù)路由
# 添加 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環(huán)境）
部署所需Dockerfile文件代碼如下
FROM python:3.7-slim
WORKDIR /app
COPY requirements.txt .
RUN apt-get update \
&& apt-get install -y ffmpeg \
&& rm -rf /var/lib/apt/lists/* \
&& pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["python", "async_io_io.py"]
部署所需requirements.txt如下
aiohttp
aiohttp-cors
ffmpeg根目錄進(jìn)行打包及啟動(dòng)
請(qǐng)求接口實(shí)現(xiàn)轉(zhuǎn)碼
文章來源：http://www.zghlxwxcb.cn/news/detail-781833.html
5. 總結(jié)
通過以上的步驟，我們成功構(gòu)建了一個(gè)流媒體服務(wù)器控制接口，可以通過Web接口實(shí)現(xiàn)對(duì)攝像頭的 RTSP 轉(zhuǎn)碼任務(wù)的動(dòng)態(tài)管理。這個(gè)接口可以集成到現(xiàn)有的流媒體服務(wù)器中，提供更多控制和管理的可能性。文章來源地址http://www.zghlxwxcb.cn/news/detail-781833.html
到了這里，關(guān)于（aiohttp-asyncio-FFmpeg-Docker-SRS）實(shí)現(xiàn)異步攝像頭轉(zhuǎn)碼服務(wù)器的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
