1、构造程序容器镜像
app.py
import subprocess
import json
import time
import multiprocessing
import socketdef check_rtmp_server(host, port, timeout=5):try:with socket.create_connection((host, port), timeout):print(f"RTMP server at {host}:{port} is available.")return Trueexcept Exception as e:print(f"RTMP server at {host}:{port} is unavailable: {e}")return Falsedef push_rtsp_to_nginx(rtsp_url, rtmp_url, transport_protocol="tcp"):host, port = "127.0.0.1", 1935 # RTMP 服务器地址和端口while True:if not check_rtmp_server(host, port):print("RTMP server not ready. Retrying in 5 seconds...")time.sleep(5)continueprint("RTMP server is ready. Waiting 5 seconds before starting the stream...")time.sleep(5) # 等待 RTMP 服务器完全准备好ffmpeg_command = ["ffmpeg","-loglevel", "quiet","-rtsp_transport", transport_protocol,"-i", rtsp_url,"-flags", "low_delay","-fflags", "nobuffer","-bufsize", "5000k","-c:v", "copy","-c:a", "copy","-an", "-f", "flv",rtmp_url]try:print(f"Attempting to push stream from {rtsp_url} to {rtmp_url} using {transport_protocol}")process = subprocess.Popen(ffmpeg_command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)for line in process.stderr:print(line.decode('utf-8').strip())process.wait()print(f"FFmpeg process exited with code {process.returncode}")except Exception as e:print(f"Error occurred: {e}")finally:if process:process.terminate()print("FFmpeg process terminated.")print("Retrying in 5 seconds...")time.sleep(5)def monitor_streams(config_path):with open(config_path, 'r') as f:config = json.load(f)processes = []for camera in config['cameras']:transport_protocol = camera.get('transport', 'tcp')p = multiprocessing.Process(target=push_rtsp_to_nginx,args=(camera['input'], camera['output'], transport_protocol))p.start()processes.append(p)try:while True:time.sleep(10)except KeyboardInterrupt:print("Stopping streams...")for p in processes:p.terminate()p.join()if __name__ == "__main__":config_path = "config.json"monitor_streams(config_path)
config.json
{"cameras": [{"id": 1,"input": "rtsp://admin:xxx@10.91.49.251:554/video1","output": "rtmp://127.0.0.1:8080/stream_1/stream_1","transport": "tcp"},{"id": 2,"input": "rtsp://admin:xxx@10.91.49.23:554/video1","output": "rtmp://127.0.0.1:8080/stream_2/stream_2","transport": "tcp"}]
}nginx.conf
worker_processes auto; # 可以根据服务器性能调整工作进程数
rtmp_auto_push on;
rtmp_auto_push_reconnect 1s; # 配置重连间隔，确保流同步快速恢复# 事件配置
events {worker_connections 2048; # 提高单个进程的最大连接数multi_accept on;use epoll; # 如果是 Linux 系统，启用 epoll 提高性能
}http {include mime.types;default_type application/octet-stream;sendfile on;tcp_nopush on; # 启用 TCP_NODELAY，减少网络包碎片tcp_nodelay on;keepalive_timeout 65; # 保持连接时间，可以根据需要调整server {listen 80;server_name localhost;location / {root html;index index.html index.htm;}# 提供 HLS 播放服务location /hls/ {types {application/vnd.apple.mpegurl m3u8;video/mp2t ts;}root /tmp; # 对应 hls_path 根目录add_header Cache-Control no-cache;add_header Access-Control-Allow-Origin *; # 允许跨域请求（如果需要）}location /stream_1/ {types {application/vnd.apple.mpegurl m3u8;video/mp2t ts;}root /tmp; # 对应 hls_path 根目录add_header Cache-Control no-cache;add_header Access-Control-Allow-Origin *; # 允许跨域请求（如果需要）}location /stream_2/ {types {application/vnd.apple.mpegurl m3u8;video/mp2t ts;}root /tmp; # 对应 hls_path 根目录add_header Cache-Control no-cache;add_header Access-Control-Allow-Origin *; # 允许跨域请求（如果需要）}location /stream_3/ {types {application/vnd.apple.mpegurl m3u8;video/mp2t ts;}root /tmp; # 对应 hls_path 根目录add_header Cache-Control no-cache;add_header Access-Control-Allow-Origin *; # 允许跨域请求（如果需要）}location /stream_4/ {types {application/vnd.apple.mpegurl m3u8;video/mp2t ts;}root /tmp; # 对应 hls_path 根目录add_header Cache-Control no-cache;add_header Access-Control-Allow-Origin *; # 允许跨域请求（如果需要）}location /stream_5/ {types {application/vnd.apple.mpegurl m3u8;video/mp2t ts;}root /tmp; # 对应 hls_path 根目录add_header Cache-Control no-cache;add_header Access-Control-Allow-Origin *; # 允许跨域请求（如果需要）}# 启用统计页面location /stat {rtmp_stat all; # 显示所有流的统计信息rtmp_stat_stylesheet stat.xsl; # 加载 XSL 样式}# 提供 `stat.xsl` 文件的静态路径location /stat.xsl {root /usr/local/nginx/html; # 你可以根据需要修改路径}error_page 500 502 503 504 /50x.html;location = /50x.html {root html;}}
}# RTMP 配置块，放在 http 配置之外
rtmp {server {listen 1935; # RTMP 推流端口chunk_size 4096; # 增加 chunk 大小application hls {live on;record off;hls on;hls_path /tmp/hls;hls_fragment 3s;}}server {listen 8080; # 另外一个 RTMP 推流端口chunk_size 4096; # 增加 chunk 大小application stream_1 {live on;record off;hls on;hls_path /tmp/stream_1;hls_fragment 3s;}application stream_2 {live on;record off;hls on;hls_path /tmp/stream_2;hls_fragment 3s;}application stream_3 {live on;record off;hls on;hls_path /tmp/stream_3;hls_fragment 3s;}application stream_4 {live on;record off;hls on;hls_path /tmp/stream_4;hls_fragment 3s;}application stream_5 {live on;record off;hls on;hls_path /tmp/stream_5;hls_fragment 3s;}}
}
entrypoint.sh
#!/bin/bash
# 启动 nginx 和其他服务
/usr/local/nginx/sbin/nginx
python3 /app/app.pydockerfile
FROM ubuntu:20.04# 设置非交互模式避免构建时的交互提示
ENV DEBIAN_FRONTEND=noninteractive# 更新源并安装依赖
RUN apt-get update && apt-get install -y \curl \unzip \build-essential \libpcre3-dev \zlib1g-dev \libssl-dev \python3-pip \ffmpeg && \apt-get clean && \rm -rf /var/lib/apt/lists/*# 创建工作目录并复制应用程序
WORKDIR /tmp# 复制本地的 NGINX 和 RTMP 模块压缩包到容器内
COPY nginx-1.27.3.tar.gz /tmp/nginx-1.27.3.tar.gz
COPY nginx-rtmp.zip /tmp/nginx-rtmp.zip# 解压并安装 NGINX 和 RTMP 模块
RUN tar zxvf nginx-1.27.3.tar.gz && \unzip nginx-rtmp.zip && \cd nginx-1.27.3 && \./configure --with-http_ssl_module --add-module=../nginx-rtmp-module-master && \make && make install && \cd .. && rm -rf nginx-1.27.3 nginx-1.27.3.tar.gz nginx-rtmp.zip nginx-rtmp-module-master# 安装 Python 库
RUN pip3 install --no-cache-dir ffmpeg-python# 设置入口脚本
COPY entrypoint.sh /entrypoint.sh
RUN chmod +x /entrypoint.sh# 创建工作目录并复制应用程序
WORKDIR /app
COPY app.py /app/app.py
COPY config.json /app/config.json# 复制 NGINX 配置文件
COPY nginx.conf /usr/local/nginx/conf/nginx.conf# 复制stat文件
COPY stat.xsl /usr/local/nginx/html/stat.xsl#设置stat.xsl文件权限
RUN chmod 644 /usr/local/nginx/html/stat.xsl# 开放端口
EXPOSE 1935 8080 80# 启动脚本，运行 NGINX 和推流服务
CMD ["/entrypoint.sh"]
另外还需要nginx和nginx-rtmp的包自行下载
构建容器
docker build -t nginx-ffmpeg-server .
2、运行
docker run -d -p 80:80 -p 1935:1935 -p 8080:8080 --name rtmp nginx-ffmpeg-server
另一种方法测试
app.py
import av
import json
import requests
import multiprocessing
import time
import xml.etree.ElementTree as ETdef fetch_rtmp_stats(url):"""从 RTMP 统计页面获取统计数据 (XML 格式)"""try:response = requests.get(url, timeout=5)if response.status_code == 200:return ET.fromstring(response.text)else:print(f"[ERROR] Failed to fetch stats. HTTP {response.status_code}")return Noneexcept Exception as e:print(f"[ERROR] Could not fetch RTMP stats: {e}")return Nonedef push_rtsp_to_rtmp(rtsp_url, rtmp_url, transport_protocol="tcp"):"""从 RTSP 拉流并直接推送到 RTMP（无需重新编码）。"""try:print(f"[INFO] Starting stream: {rtsp_url} -> {rtmp_url}")input_container = av.open(rtsp_url,options={"rtsp_transport": transport_protocol, "timeout": "5000000"})output_container = av.open(rtmp_url, mode="w", format="flv")video_stream = Noneaudio_stream = Nonefor stream in input_container.streams:if stream.type == "video" and not video_stream:video_stream = output_container.add_stream(template=stream)elif stream.type == "audio" and not audio_stream:audio_stream = output_container.add_stream(template=stream)for packet in input_container.demux(video_stream, audio_stream):if packet.dts is None:continuepacket.stream = video_stream if packet.stream.type == "video" else audio_streamoutput_container.mux(packet)except Exception as e:print(f"[ERROR] Failed to push stream: {rtsp_url} -> {rtmp_url}: {e}")raise # 重新抛出异常，通知调用方进程失败finally:if 'input_container' in locals():input_container.close()if 'output_container' in locals():output_container.close()print(f"[INFO] Stopped pushing stream: {rtsp_url} -> {rtmp_url}")def check_stream_status(stats_url, stream_name, retries=3, check_interval=5):"""检查指定流的状态，最多尝试 `retries` 次，返回流是否正常"""failure_count = 0for _ in range(retries):stats = fetch_rtmp_stats(stats_url)if stats is None:print("[WARNING] Could not fetch RTMP stats. Retrying...")failure_count += 1else:# 查找当前流状态stream_nodes = stats.findall(f".//stream[name='{stream_name}']")is_active = stream_nodes and all(node.find("active") is not None for node in stream_nodes)if is_active:print(f"[INFO] Stream {stream_name} is active.")return True # 流正常else:print(f"[WARNING] Stream {stream_name} is inactive.")failure_count += 1time.sleep(check_interval) # 等待几秒后再次尝试# 如果检查了所有次数后都失败了，才认为流异常if failure_count == retries:print(f"[ERROR] Stream {stream_name} is inactive after {retries} checks.")return False # 流不正常return True # 如果至少有一次成功，则认为流正常def monitor_streams(config_path, stats_url, check_interval=60):"""定期从 /stat 页面获取流状态，如果流异常则尝试恢复已有推流进程，不成功则重新启动新推流进程。"""with open(config_path, "r") as f:config = json.load(f)processes = {}# 初始化：先启动所有流的推流进程for camera in config.get("cameras", []):stream_name = camera["stream"]rtsp_url = camera["input"]rtmp_url = camera["output"]transport_protocol = camera.get("transport", "tcp")print(f"[INFO] Starting initial stream for {stream_name}...")new_proc = multiprocessing.Process(target=push_rtsp_to_rtmp, args=(rtsp_url, rtmp_url, transport_protocol))new_proc.start()processes[stream_name] = new_proc# 定期检查流状态while True:for camera in config.get("cameras", []):stream_name = camera["stream"]rtsp_url = camera["input"]rtmp_url = camera["output"]transport_protocol = camera.get("transport", "tcp")# 检查流状态，如果流异常，则进行处理print(f"[INFO] Checking status of stream {stream_name}...")if not check_stream_status(stats_url, stream_name):print(f"[WARNING] Stream {stream_name} is inactive. Attempting recovery...")# 如果进程存在并且存活，尝试恢复推流if stream_name in processes:proc = processes[stream_name]if proc.is_alive():print(f"[INFO] Attempting to recover existing process for stream {stream_name}.")try:# 尝试恢复推流并再次检查流状态proc.join(timeout=1) # 检查现有进程的状态except Exception as e:print(f"[ERROR] Existing process for stream {stream_name} failed: {e}")proc.terminate() # 终止失败的进程proc.join()print(f"[INFO] Terminated failed process for stream {stream_name}.")proc = None# 再检查两次流状态，如果流仍然异常，则重新启动新进程recovery_attempts = 2for _ in range(recovery_attempts):if check_stream_status(stats_url, stream_name):print(f"[INFO] Stream {stream_name} has recovered.")breakprint(f"[WARNING] Stream {stream_name} is still inactive after recovery attempt.")time.sleep(5)# 如果仍然失败，则重新启动进程if not check_stream_status(stats_url, stream_name):print(f"[ERROR] Stream {stream_name} could not be recovered. Restarting...")proc = Noneelse:print(f"[INFO] Existing process for stream {stream_name} is not alive. Restarting.")proc = Noneelse:proc = None# 如果没有存活的进程或进程不可用，启动新的推流进程if proc is None:print(f"[INFO] Starting a new process for stream {stream_name}.")new_proc = multiprocessing.Process(target=push_rtsp_to_rtmp, args=(rtsp_url, rtmp_url, transport_protocol))new_proc.start()processes[stream_name] = new_procelse:print(f"[INFO] Stream {stream_name} is healthy. No action needed.")time.sleep(check_interval)if __name__ == "__main__":CONFIG_PATH = "config.json"STATS_URL = "http://127.0.0.1/stat"monitor_streams(CONFIG_PATH, STATS_URL)
config.json
{"cameras": [{"id": 1,"stream": "stream_1","input": "rtsp://admin:xxx@10.91.49.251:554/video1?rtsp_transport=tcp","output": "rtmp://127.0.0.1:8080/stream_1/stream_1","transport": "tcp"},{"id": 2,"stream": "stream_2","input": "rtsp://admin:xxx@10.91.49.23:554/video1?rtsp_transport=tcp","output": "rtmp://127.0.0.1:8080/stream_2/stream_2","transport": "tcp"}]}
entrypoint.sh
#!/bin/bash
# 启动 NGINX
/usr/local/nginx/sbin/nginx# 启动 Python 推流服务
python3 /app/app.py
nginx.conf
worker_processes auto; # 可以根据服务器性能调整工作进程数
rtmp_auto_push on;
rtmp_auto_push_reconnect 1s; # 配置重连间隔，确保流同步快速恢复# 事件配置
events {worker_connections 2048; # 提高单个进程的最大连接数multi_accept on;use epoll; # 如果是 Linux 系统，启用 epoll 提高性能
}http {include mime.types;default_type application/octet-stream;sendfile on;tcp_nopush on; # 启用 TCP_NODELAY，减少网络包碎片tcp_nodelay on;keepalive_timeout 65; # 保持连接时间，可以根据需要调整server {listen 80;server_name localhost;location / {root html;index index.html index.htm;}# 提供 HLS 播放服务location /hls/ {types {application/vnd.apple.mpegurl m3u8;video/mp2t ts;}root /tmp; # 对应 hls_path 根目录add_header Cache-Control no-cache;add_header Access-Control-Allow-Origin *; # 允许跨域请求（如果需要）}location /stream_1/ {types {application/vnd.apple.mpegurl m3u8;video/mp2t ts;}root /tmp; # 对应 hls_path 根目录add_header Cache-Control no-cache;add_header Access-Control-Allow-Origin *; # 允许跨域请求（如果需要）}location /stream_2/ {types {application/vnd.apple.mpegurl m3u8;video/mp2t ts;}root /tmp; # 对应 hls_path 根目录add_header Cache-Control no-cache;add_header Access-Control-Allow-Origin *; # 允许跨域请求（如果需要）}location /stream_3/ {types {application/vnd.apple.mpegurl m3u8;video/mp2t ts;}root /tmp; # 对应 hls_path 根目录add_header Cache-Control no-cache;add_header Access-Control-Allow-Origin *; # 允许跨域请求（如果需要）}location /stream_4/ {types {application/vnd.apple.mpegurl m3u8;video/mp2t ts;}root /tmp; # 对应 hls_path 根目录add_header Cache-Control no-cache;add_header Access-Control-Allow-Origin *; # 允许跨域请求（如果需要）}location /stream_5/ {types {application/vnd.apple.mpegurl m3u8;video/mp2t ts;}root /tmp; # 对应 hls_path 根目录add_header Cache-Control no-cache;add_header Access-Control-Allow-Origin *; # 允许跨域请求（如果需要）}# 启用统计页面location /stat {rtmp_stat all; # 显示所有流的统计信息rtmp_stat_stylesheet stat.xsl; # 加载 XSL 样式}# 提供 `stat.xsl` 文件的静态路径location /stat.xsl {root /usr/local/nginx/html; # 你可以根据需要修改路径}error_page 500 502 503 504 /50x.html;location = /50x.html {root html;}}
}# RTMP 配置块，放在 http 配置之外
rtmp {server {listen 1935; # RTMP 推流端口chunk_size 4096; # 增加 chunk 大小application hls {live on;record off;hls on;hls_path /tmp/hls;hls_fragment 3s;}}server {listen 8080; # 另外一个 RTMP 推流端口chunk_size 4096; # 增加 chunk 大小application stream_1 {live on;record off;hls on;hls_path /tmp/stream_1;hls_fragment 3s;}application stream_2 {live on;record off;hls on;hls_path /tmp/stream_2;hls_fragment 3s;}application stream_3 {live on;record off;hls on;hls_path /tmp/stream_3;hls_fragment 3s;}application stream_4 {live on;record off;hls on;hls_path /tmp/stream_4;hls_fragment 3s;}application stream_5 {live on;record off;hls on;hls_path /tmp/stream_5;hls_fragment 3s;}}
}
dockerfile
FROM ubuntu:20.04# 设置非交互模式，避免构建时的交互提示
ENV DEBIAN_FRONTEND=noninteractive# 更新源并安装依赖
RUN apt-get update && apt-get install -y \curl \unzip \build-essential \libpcre3-dev \zlib1g-dev \libssl-dev \python3-pip \python3-dev \pkg-config \libavformat-dev \libavcodec-dev \libavdevice-dev \libavutil-dev \libswscale-dev \libswresample-dev \libopencv-dev && \apt-get clean && \rm -rf /var/lib/apt/lists/*# 创建工作目录并复制 NGINX 和 RTMP 模块压缩包到容器内
WORKDIR /tmp
COPY nginx-1.27.3.tar.gz /tmp/nginx-1.27.3.tar.gz
COPY nginx-rtmp.zip /tmp/nginx-rtmp.zip# 解压并安装 NGINX 和 RTMP 模块
RUN tar zxvf nginx-1.27.3.tar.gz && \unzip nginx-rtmp.zip && \cd nginx-1.27.3 && \./configure --with-http_ssl_module --add-module=../nginx-rtmp-module-master && \make && make install && \cd .. && rm -rf nginx-1.27.3 nginx-1.27.3.tar.gz nginx-rtmp.zip nginx-rtmp-module-master# 安装 Python 库
RUN pip3 install --no-cache-dir \opencv-python-headless \av \requests \numpy# 设置入口脚本
COPY entrypoint.sh /entrypoint.sh
RUN chmod +x /entrypoint.sh# 创建工作目录并复制应用程序
WORKDIR /app
COPY app.py /app/app.py
COPY config.json /app/config.json# 复制 NGINX 配置文件
COPY nginx.conf /usr/local/nginx/conf/nginx.conf# 复制 stat 文件
COPY stat.xsl /usr/local/nginx/html/stat.xsl# 设置 stat.xsl 文件权限
RUN chmod 644 /usr/local/nginx/html/stat.xsl# 开放端口
EXPOSE 1935 8080 80# 启动脚本，运行 NGINX 和推流服务
CMD ["/entrypoint.sh"]