1. 多路摄像头图像采集的架构选型与核心思路多路摄像头图像采集这件事看起来只是读图存图两个动作但真正落到工程里尤其是路数一多、分辨率一高问题就会集中爆发。我最早接触这类需求是在一个模拟的视觉分析项目里当时需要同时接入8路摄像头每路1080p、25帧要求把原始帧完整落盘供后续的算法模块离线跑。第一版代码写得很直白——一个循环里挨个read、挨个imwrite结果跑起来帧率直接掉到个位数丢帧丢到怀疑人生。后来逐步迭代才形成了标题里这套多线程读取多进程保存的组合拳。这套架构的核心思想其实一句话就能说清把I/O密集型的读取和CPU/磁盘密集型的编码保存拆开用不同的并发模型分别处理中间用队列解耦。为什么读取用多线程、保存用多进程这是整个方案里最关键的选型决策下面展开讲。1.1 为什么读取用多线程而不是多进程摄像头读取无论是USB摄像头、RTSP网络流还是工业相机SDK本质上是I/O等待密集型任务。以RTSP流为例read()调用的大部分时间都花在等待网络数据包到达上CPU几乎不干活。这种场景下多线程是天然合适的线程创建和切换开销远小于进程8路、16路摄像头开对应数量的读取线程系统负担很小。线程之间共享内存读取到的帧可以直接放进同一个队列不需要跨进程序列化省掉了大量拷贝和IPC开销。Python里虽然有GIL但摄像头read()这类底层调用在等待I/O时会释放GIL所以多线程读取确实能并行起来。我实测过用多进程去读RTSP流进程间要通过管道传图像数据一帧1080p的原始数据是1920×1080×3≈6MB序列化和传输的代价比读取本身还大完全得不偿失。所以读取环节多线程是更优解。1.2 为什么保存用多进程而不是多线程保存环节就不一样了。图像编码比如JPEG压缩和写磁盘是实打实的CPU和I/O消耗。JPEG编码是纯计算任务会长时间持有GIL这时候多线程就废了——所有线程排队等GIL等于串行执行。而多进程每个进程有独立的Python解释器和GIL能真正并行利用多核CPU。另一个原因是磁盘写入。多路摄像头同时写盘如果都用线程磁盘I/O调度会非常混乱而且Python的GIL会让写操作互相阻塞。用多进程每个保存进程独立处理自己的写队列配合操作系统的I/O调度吞吐量能提升好几倍。提示这里有个容易踩的坑——保存进程数量不是越多越好。我试过开16个保存进程对应16路摄像头结果磁盘随机写严重反而比8个进程慢。一般建议保存进程数控制在CPU核心数的1到1.5倍具体要看磁盘类型SSD还是机械盘。1.3 整体数据流与队列设计整套架构的数据流是这样的每个摄像头对应一个读取线程读取线程把帧放进一个共享的帧队列保存进程从队列里取帧编码后写盘。这里队列的选择很关键如果读取线程和保存进程在同一个进程内可以用queue.Queue但多进程场景下必须用multiprocessing.Queue。multiprocessing.Queue底层基于管道和锁跨进程传递大对象比如6MB的numpy数组会有序列化开销。我的做法是在读取线程里就把帧编码成JPEG字节流这样队列里传的是压缩后的小数据通常几百KB序列化代价大幅降低。这个读取线程内先编码的优化是我踩了很多坑才总结出来的。一开始我传原始帧队列内存暴涨程序跑几分钟就OOM。改成传编码后的字节流内存占用直接降了一个数量级。2. 核心细节解析与实操要点架构定下来之后真正决定成败的是细节。这一块我分几个关键点来讲都是实际跑起来才会暴露的问题。2.1 帧队列的容量与背压控制队列不能无限大否则内存会被撑爆也不能太小否则读取线程频繁阻塞导致摄像头缓冲区溢出丢帧。我的经验值是队列容量设置为保存进程数×2到保存进程数×4之间。举个例子8个保存进程队列容量设16到32比较合适。这样读取线程偶尔快一点不会立刻阻塞保存进程偶尔慢一点也有缓冲空间。如果队列满了读取线程的put操作会阻塞这时候摄像头的帧会堆积在驱动层缓冲区最终导致丢帧——这是正常的背压机制说明保存环节跟不上需要优化保存性能或者减少路数。注意multiprocessing.Queue的put默认是阻塞的但如果你用put_nowait队列满时会抛异常。我建议用阻塞put配合超时比如put(frame, timeout1)超时后记录日志并丢弃当前帧避免整个读取线程卡死。2.2 摄像头读取的异常处理与重连多路摄像头最烦人的就是某一路突然断流。RTSP流可能因为网络抖动断开USB摄像头可能因为供电不稳掉线。如果读取线程没有异常处理一路挂了可能导致整个程序崩溃或者那一路永久卡死。我的做法是在每个读取线程里加重连循环def capture_worker(cam_id, rtsp_url, frame_queue, stop_event): while not stop_event.is_set(): cap cv2.VideoCapture(rtsp_url) if not cap.isOpened(): time.sleep(2) continue while not stop_event.is_set(): ret, frame cap.read() if not ret: break # 跳出内层循环触发重连 # 编码后放入队列 _, jpg cv2.imencode(.jpg, frame, [cv2.IMWRITE_JPEG_QUALITY, 85]) try: frame_queue.put((cam_id, jpg.tobytes()), timeout1) except queue.Full: pass # 丢弃当前帧 cap.release() time.sleep(1) # 重连前等待这段代码的关键点是ret为False时跳出内层循环释放摄像头等待1秒后重新打开。这样即使某一路断流其他路不受影响断流的那一路会自动恢复。2.3 保存进程的编码参数与文件命名保存进程拿到的是JPEG字节流直接写文件即可不需要再编码。但文件命名和目录组织有讲究按摄像头ID分目录output/cam_0/、output/cam_1/避免单目录文件过多导致文件系统性能下降。文件名带时间戳和帧序号20250101_120000_000123.jpg方便后续按时间检索。定期切换文件或目录如果连续跑几小时单目录会有几十万文件建议每小时切一个子目录。JPEG质量参数我一般设85这是质量和体积的平衡点。设95体积会翻倍但肉眼几乎看不出差别设70会有明显块状伪影。如果是做算法分析的原始帧建议存PNG或者无损格式但体积会大很多需要权衡。2.4 进程间共享状态的清理多进程程序退出时如果队列里还有数据没消费完直接terminate会导致数据丢失。正确的退出流程是设置stop_event通知所有读取线程停止。等待读取线程join完成。向队列发送哨兵值比如None通知保存进程退出。等待保存进程join完成。最后关闭队列释放资源。我见过太多程序因为退出流程没写好导致僵尸进程残留或者文件写了一半损坏。这个流程虽然啰嗦但必须严格执行。3. 完整实操过程与核心环节实现下面我把整套代码拆成可运行的模块你可以直接参考复现。环境是Python 3.8依赖opencv-python和标准库的multiprocessing、threading、queue。3.1 环境准备与依赖安装pip install opencv-python numpy如果是RTSP流OpenCV需要带FFmpeg支持官方预编译的opencv-python包已经包含。工业相机的话需要装厂商SDK这里以通用USB摄像头和RTSP为例。3.2 主控程序与参数配置先定义配置区把所有可调参数集中管理import cv2 import os import time import queue import threading import multiprocessing as mp from datetime import datetime # 配置区 CAMERAS [ {id: 0, source: 0}, # USB摄像头 {id: 1, source: rtsp://example/stream1}, # RTSP流 {id: 2, source: rtsp://example/stream2}, # ... 按需添加 ] OUTPUT_DIR ./output QUEUE_SIZE 32 SAVE_PROCESSES 4 JPEG_QUALITY 85这里SAVE_PROCESSES设4是因为我的测试机是8核留一半给读取线程和系统。如果你的机器核多可以适当增加但别超过核数。3.3 读取线程的实现读取线程负责拉流、编码、入队def capture_worker(cam_id, source, frame_queue, stop_event): os.makedirs(os.path.join(OUTPUT_DIR, fcam_{cam_id}), exist_okTrue) while not stop_event.is_set(): cap cv2.VideoCapture(source) if not cap.isOpened(): print(f[cam_{cam_id}] 打开失败2秒后重试) time.sleep(2) continue print(f[cam_{cam_id}] 已连接) while not stop_event.is_set(): ret, frame cap.read() if not ret: print(f[cam_{cam_id}] 读取失败准备重连) break ok, jpg cv2.imencode(.jpg, frame, [cv2.IMWRITE_JPEG_QUALITY, JPEG_QUALITY]) if not ok: continue try: frame_queue.put((cam_id, jpg.tobytes()), timeout1) except queue.Full: pass # 队列满丢弃当前帧 cap.release() time.sleep(1)注意cv2.imencode返回的是(retval, buf)buf是numpy数组要.tobytes()转成bytes才能跨进程传递。3.4 保存进程的实现保存进程从队列取数据写文件def save_worker(frame_queue, stop_event): while not stop_event.is_set(): try: item frame_queue.get(timeout1) except queue.Empty: continue if item is None: # 哨兵退出 break cam_id, jpg_bytes item now datetime.now() ts now.strftime(%Y%m%d_%H%M%S_%f)[:-3] dir_path os.path.join(OUTPUT_DIR, fcam_{cam_id}, now.strftime(%Y%m%d_%H)) os.makedirs(dir_path, exist_okTrue) file_path os.path.join(dir_path, f{ts}.jpg) with open(file_path, wb) as f: f.write(jpg_bytes)这里按小时切目录避免单目录文件过多。时间戳精确到毫秒基本不会重名。3.5 主流程组装把线程和进程组装起来def main(): mp.set_start_method(spawn) # 跨平台兼容 frame_queue mp.Queue(maxsizeQUEUE_SIZE) stop_event mp.Event() # 启动保存进程 save_procs [] for _ in range(SAVE_PROCESSES): p mp.Process(targetsave_worker, args(frame_queue, stop_event)) p.start() save_procs.append(p) # 启动读取线程 read_threads [] for cam in CAMERAS: t threading.Thread(targetcapture_worker, args(cam[id], cam[source], frame_queue, stop_event)) t.start() read_threads.append(t) try: while True: time.sleep(1) except KeyboardInterrupt: print(收到退出信号正在清理...) stop_event.set() # 等待读取线程结束 for t in read_threads: t.join(timeout5) # 发送哨兵通知保存进程退出 for _ in save_procs: frame_queue.put(None) for p in save_procs: p.join(timeout5) if p.is_alive(): p.terminate() print(已安全退出) if __name__ __main__: main()mp.set_start_method(spawn)在Windows上必须Linux上默认fork也可以但spawn更安全避免fork带来的锁继承问题。3.6 参数计算与性能预估假设8路1080p、25帧JPEG质量85单帧压缩后约200KB。总数据率是8路 × 25帧/秒 × 200KB 40MB/s这个写入速率机械硬盘顺序写约100MB/s勉强够SSD500MB/s很轻松。如果路数翻倍到16路就是80MB/s机械盘就吃力了建议上SSD或者做RAID。CPU方面JPEG编码单帧1080p约5-10ms取决于CPU8路25帧就是200帧/秒需要200×7.5ms≈1.5秒的CPU时间分摊到多个核心上4核基本够用。如果编码成为瓶颈可以降低JPEG质量或者用硬件编码如NVENC。4. 常见问题与排查技巧实录这套架构跑起来之后我遇到过不少问题这里整理成速查表方便你对照排查。4.1 常见问题速查表问题现象可能原因排查方法解决方案帧率远低于预期保存进程不足或磁盘慢看队列是否长期满增加保存进程或换SSD内存持续增长队列积压或帧未释放监控队列大小和内存减小队列容量检查编码某一路频繁断流网络抖动或摄像头问题看该路重连日志增加重连等待检查网络程序退出后进程残留退出流程不完整ps查看残留进程严格按哨兵流程退出文件写了一半损坏进程被强杀检查文件大小用临时文件重命名CPU占用100%但吞吐低GIL争用看是否保存用了线程保存必须用多进程4.2 独家避坑技巧技巧一用临时文件原子重命名避免损坏文件。写文件时先写.tmp写完再os.rename成.jpg。这样即使进程中途被杀也不会留下损坏的jpg文件后续处理程序不会读到半截数据。技巧二读取线程里做编码而不是保存进程里做。这个前面提过但值得再强调。读取线程本来就在等I/OCPU闲着顺手把编码做了保存进程就只剩纯写盘吞吐量能提升30%以上。技巧三用spawn而不是fork启动进程。在Linux上fork虽然快但如果父进程里有线程比如OpenCV内部线程fork后子进程可能死锁。spawn虽然慢一点但稳定得多尤其是程序要长时间运行。技巧四监控队列深度动态调整。我在生产环境加了一个监控线程每秒打印一次队列深度。如果队列长期接近满说明保存跟不上可以动态增加保存进程如果长期为空说明读取慢可以检查摄像头。这个监控帮我定位过好几次性能瓶颈。技巧五时间戳用单调时钟。datetime.now()会受系统时间调整影响如果NTP校时导致时间跳变文件名可能乱序。可以用time.monotonic()做帧序号配合启动时间换算成绝对时间。4.3 性能调优的进阶思路如果基础版本跑通了还想进一步压榨性能有几个方向零拷贝传输用multiprocessing.shared_memory共享内存传帧避免队列序列化。但实现复杂需要自己管理内存生命周期适合帧率极高的场景。硬件编码用GPU做JPEG编码CPU占用能降一个数量级。OpenCV的cudacodec模块支持但需要编译带CUDA的版本。异步写盘用aiofiles或者io_uring做异步写进一步降低写盘延迟。不过Python的异步文件IO生态还不成熟收益有限。分级存储热数据写SSD冷数据定期迁移到机械盘。适合长时间运行的场景。我个人在实际操作中的体会是先把基础版本跑稳再谈优化。很多性能问题不是架构问题而是某个参数没调对比如队列太小、保存进程太多、JPEG质量太高。基础版本用监控数据说话找到真正的瓶颈再针对性优化比一上来就上复杂方案靠谱得多。最后再分享一个小技巧如果你的摄像头支持MJPEG输出很多USB摄像头和RTSP流都支持可以直接拿MJPEG流省掉自己编码的步骤读取线程只负责拆包和入队CPU占用能再降一截。这个在OpenCV里可以通过设置cv2.CAP_PROP_FOURCC为MJPG来启用实测下来很稳。