Axis IP摄像机视频流广播时出现async_lock错误求助

首先，你遇到的async_lock错误大概率是因为异步锁的使用场景不对——比如在同步函数里尝试获取异步锁，或者锁的初始化、释放逻辑有问题。咱们直接换个更可靠的实现思路：用aiohttp搭建异步HTTP服务器，配合ffmpeg拉取并转码摄像机流，同时用异步锁保证只有一个连接去拉取摄像机的源流，所有客户端共享转码后的流，这样就能实现单连接多用户预览的需求。

准备工作 #

先确保你的服务器上安装好依赖：

• Python的aiohttp库：pip install aiohttp
• ffmpeg：根据系统安装（比如Ubuntu用apt install ffmpeg，CentOS用yum install ffmpeg）

完整实现代码 #

import asyncio
import aiohttp
from aiohttp import web
import subprocess
from typing import Optional

# 摄像机的RTSP地址（RTSP比MJPG传输效率更高）
RTSP_URL = "rtsp://ca.me.ra.ip:554/axis-media/media.amp?videocodec=h264&resolution=640x480"
# 异步锁，保证同一时间只有一个进程拉取摄像机源流
stream_lock = asyncio.Lock()
# 存储ffmpeg的输出流，供所有客户端共享
stream_reader: Optional[asyncio.StreamReader] = None

async def start_ffmpeg_stream():
    global stream_reader
    # 用ffmpeg把RTSP流转成HTTP可直接播放的MPEG-TS格式
    cmd = [
        "ffmpeg",
        "-rtsp_transport", "tcp",  # 用TCP传输RTSP，避免丢包卡顿
        "-i", RTSP_URL,
        "-c:v", "copy",  # 直接复制H.264编码，减少服务器资源消耗
        "-f", "mpegts",  # 输出MPEG-TS格式，适配绝大多数播放器
        "-"  # 输出到标准输出，供后续读取
    ]
    # 启动ffmpeg异步进程
    process = await asyncio.create_subprocess_exec(
        *cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=subprocess.DEVNULL  # 忽略错误输出，调试时可改成文件路径查看日志
    )
    stream_reader = process.stdout
    # 等待进程结束，若意外退出则重置流状态
    await process.wait()
    stream_reader = None

async def stream_handler(request):
    global stream_reader
    # 用异步锁控制源流的启动逻辑
    async with stream_lock:
        if stream_reader is None:
            # 没有活跃流时，启动ffmpeg拉取
            asyncio.create_task(start_ffmpeg_stream())
            # 等待流初始化完成
            while stream_reader is None:
                await asyncio.sleep(0.1)
    
    # 设置响应头，告诉客户端这是视频流
    headers = {
        "Content-Type": "video/MP2T",
        "Cache-Control": "no-cache",
        "Connection": "close"
    }
    response = web.StreamResponse(headers=headers)
    await response.prepare(request)
    
    try:
        # 持续读取ffmpeg输出，分块发送给客户端
        while not stream_reader.at_eof():
            chunk = await stream_reader.read(4096)
            if not chunk:
                break
            await response.write(chunk)
    except Exception as e:
        print(f"客户端连接异常: {e}")
    finally:
        await response.done()
    return response

async def main():
    app = web.Application()
    app.add_routes([web.get("/stream", stream_handler)])
    # 启动服务器，监听所有网卡的8080端口
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, "0.0.0.0", 8080)
    await site.start()
    print("服务器已启动，访问http://你的服务器IP:8080/stream即可预览")
    # 保持服务器持续运行
    await asyncio.Event().wait()

if __name__ == "__main__":
    asyncio.run(main())

代码说明 #

1. 异步锁的正确用法：stream_lock确保同一时间只有一个逻辑去启动ffmpeg进程，避免重复拉取摄像机的源流，彻底解决你之前遇到的async_lock错误。
2. 高效流处理：用RTSP源流转码成MPEG-TS格式，比MJPG流的传输效率高很多，同时支持浏览器直接播放，无需额外插件。
3. 共享流逻辑：所有客户端连接/stream路径时，都会读取同一个ffmpeg输出流，真正实现“单连接拉取、多用户共享”的需求。
4. 容错机制：如果ffmpeg进程意外退出，会自动重置流状态，下次有客户端连接时会重新启动拉流，保证服务可用性。

测试方法 #

1. 运行代码后，在浏览器或视频播放器中访问http://你的服务器IP:8080/stream，就能看到摄像机的实时视频。
2. 打开多个浏览器窗口或设备访问同一个地址，服务器只会维持一个到AXIS摄像机的RTSP连接，所有客户端共享这个流。

如果你偏好使用MJPG源流，只需要把RTSP_URL替换成你的HTTP MJPG地址，并调整ffmpeg参数（比如添加-f mjpeg指定输入格式），不过还是更推荐用RTSP源，传输效率和稳定性都更好。

内容的提问来源于stack exchange，提问作者Ghasem