FFmpeg学习笔记(二):多线程rtsp推流和ffplay拉流操作，并储存为多路avi格式的视频

多线程

import threading
import time
# acquire the face iou
def get_frame():
    print("当前线程的信息:", threading.current_thread())
    time.sleep(1)

# acquire the face feature pkl
def get_feature():
    print("当前线程的信息:", threading.current_thread())
    time.sleep(10)

def main():
    # start the multiprocess
    t1=threading.Thread(target=get_frame,name='face_iou_save')
    t2=threading.Thread(target=get_feature,name='face_feature_pkl')
    t1.daemon = True
    t2.daemon = True # assist the process
    t1.start()
    t2.start() # start the process
    print('threading.active')
    t1.join()
    t2.join()
    print('threading.end')
if __name__ == '__main__':
    main()

如果不了解什么是rtsp推流和ffplay拉流可以看我这篇博客
https://developer.aliyun.com/article/1625071
由于项目需要，不得不使用多线程的方式来进行视频流的推送，从边缘端储存到服务器端。多线程的方式很明显有个非常大的特点，线程之间不相互影响，也就是你有你的工作，我有我的工作，你不工作了没关系，也不会影响我。
下面请看推流和拉流的代码

多进程rtsp推流

import cv2
import time
import multiprocessing as mp

def recovery_stream(cap,rtsp_url,retry_delay=10):
    while True:
        try:
            print(f"正在重新连接RTSP流:{rtsp_url}...")
            cap.release()
            cap = cv2.VideoCapture(rtsp_url)
            if cap.isOpened():
                print(f"RTSP流恢复成功:{rtsp_url}")
                return cap
        except Exception as e:
            print(f"尝试恢复RTSP流时遇到错误:{e}")
        time.sleep(retry_delay) # 等待10秒

def image_put(q, rtsp_url, rtsp_name):
    # 创建VideoCapture对象，指定RTSP流地址
    cap = cv2.VideoCapture(rtsp_url)
    fps = cap.get(cv2.CAP_PROP_FPS)
    print("FPS:", fps)
    if not cap.isOpened():
        print(f"无法打开RTSP流:{rtsp_url}")
        cap = recovery_stream(cap,rtsp_url,retry_delay=10)

    while True:
        ret, frame = cap.read()
        if not ret:
            cap = recovery_stream(cap,rtsp_url,retry_delay=10)
            ret,frame = cap.read()

        while q.qsize() > 1:
            try:
                _ = q.get_nowait()
            except Exception as e:
                pass

        q.put(frame)
        time.sleep(1/fps)
        # time.sleep(0.01)

def image_get(q, rtsp_url, rtsp_name):
    windowname = rtsp_url
    cv2.namedWindow(windowname)
    i = 0
    while True:
        frame = q.get(block=True)
        resized_frame = cv2.resize(frame, (640, 480))
        cv2.imshow(windowname, resized_frame)
        i += 1
        if cv2.waitKey(25) == 27: 
            break
        print("{}:{}".format(rtsp_name,i))

def run_multi_camera(rtsp_urls, rtsp_names):
    mp.set_start_method('spawn')
    queues = [mp.Queue(maxsize=2) for _ in rtsp_urls]
    queues1 = [mp.Queue(maxsize=2) for _ in rtsp_urls] # 备用
    stop_event = mp.Event()  # 创建一个停止事件
    processes = []

    for queue,rtsp_url,rtsp_name in zip(queues,rtsp_urls,rtsp_names):
        processes.append(mp.Process(target=image_put, args=(queue, rtsp_url, rtsp_name)))
        processes.append(mp.Process(target=image_get, args=(queue, rtsp_url, rtsp_name)))

    for process in processes:
        process.daemon = True
        process.start()
    for process in processes:
        process.join()
    stop_event.set()  # 设置停止事件，通知所有子进程退出

if __name__ == '__main__':
    # RTSP视频流地址
    rtsp_urls = ['rtsp://21*******e442aae',
            'rtsp://21**********d14275']
    rtsp_names = ["video1", "video2"]
    run_multi_camera(rtsp_urls, rtsp_names)

# 代码描述：利用多进程方法，利用两个海康威视摄像头，同时录取视频并保存本地

import cv2
import time
import multiprocessing as mp
import subprocess as sp
import  traceback

num = 0

# 抓取图片，确认视频流的读入
def image_put(q, name, pwd, ip, channel,ids):
    # if type(channel)== int:
    #cv2.namedWindow(ip, cv2.WINDOW_NORMAL)
    global url
    url="rtsp://%s:%s@%s:%s//Streaming/Channels/%s" \
                           % (name, pwd, ip, channel,ids)
    cap = cv2.VideoCapture(url)
    # 获取视频帧率
    fps = cap.get(cv2.CAP_PROP_FPS)
    print('fps: ', fps)
    if cap.isOpened():
        print('HIKVISION1')
        print('camera ' + ip + " connected.")
    #else:
    #    cap = cv2.VideoCapture("rtsp://%s:%s@%s/cam/realmonitor?channel=%d&subtype=0" % (name, pwd, ip, channel))
    #   print('DaHua')
    while cap.isOpened():
        # print('cap.read()[0]:', cap.read()[0])
        ret, frame = cap.read()
        # print('ret:', ret)
        #frame = cv2.resize(frame, (800, 600))
        # 抓取图片不成功再重新抓取
        if not ret:
            cap = cv2.VideoCapture("rtsp://%s:%s@%s:%s//Streaming/Channels/1" \
                                   % (name, pwd, ip, channel))
            print('HIKVISION2')
            ret, frame = cap.read()
            #frame = cv2.resize(frame, (800,600))
        # Press esc on keyboard to  exit
        # if cv2.waitKey(1) & 0xFF == 27:
        #     break
        q.put(frame) # 线程A不仅将图片放入队列
        # print('q.qsize():',(q.qsize() > 1))
        q.get() if q.qsize() > 1 else time.sleep(0.01) # 线程A还负责移除队列中的旧图
    cap.release()

# 获得视频流帧数图片，保存读入的视频
def image_get(q, name, pwd, ip, channel,ids,command):
    while True:
        if len(command) > 0:
            # 管道配置
            pipe = sp.Popen(command,shell=True, stdin=sp.PIPE)# ,shell=False
            break
    if pipe.poll() is not None:
        print(pipe.poll())
        #pipe = sp.Popen(command,shell=True, stdin=sp.PIPE)# ,shell=False
        time.sleep(3)
    while True:
        num += 1
        frame = q.get()
        if num == 100:
            print("start sleep 50")
            time.sleep(50)
            print("end sleep 50")
        if num >= 100:
            print("超时后的，第%d次写入" % (num - 99))
            time.sleep(5)
        try:
            pipe.stdin.write(frame.tostring())  # 存入管道用于直播
        except:
            traceback.print_exc()

# 解决进程问题
def run_multi_camera():
    # user_name, user_pwd = "admin", "password"
    user_name, user_pwd = "admin", "a12345678"
    # 摄像头的账户密码改成自己摄像头注册的信息
    camera_ip_l = [
         # ipv4
        "10.16.55.149",
        "10.16.55.150",
        "10.16.55.151",
        "10.16.55.152",
        # 把你的摄像头的地址放到这里，如果是ipv6，那么需要加一个中括号
    ]
    ports = ['554', '555', '556','557']#'554',
    idss=['1','2','3','4']
    commands=[]
    for camera_ip,port,ids in zip(camera_ip_l,ports,idss):
        command="""ffmpeg -re -rtsp_transport tcp -i \"rtsp://admin:a12345678@{}:{}//Streaming/Channels/1\" -f flv -vcodec libx264 -vprofile baseline -acodec aac -ar 44100 -strict -2 -ac 1 -f flv -flvflags no_duration_filesize -r 29.97 -s 1280x720 -q 10 \"rtmp://127.0.0.1:1935/live/{}""".format(camera_ip,port,ids)
        command=command+'"'
        commands.append(command)
    mp.set_start_method(method='spawn')  # init
    queues = [mp.Queue(maxsize=2) for _ in camera_ip_l]

    processes = []
    for queue, camera_ip,port,ids,command in zip(queues, camera_ip_l,ports,idss,commands):

        processes.append(mp.Process(target=image_put, args=(queue, user_name, user_pwd, camera_ip,port,ids)))
        processes.append(mp.Process(target=image_get, args=(queue, user_name, user_pwd, camera_ip,port,ids,command)))

    for process in processes:
        process.daemon = True
        process.start()
    for process in processes:
        process.join()

if __name__ == '__main__':
    run_multi_camera()           # 调用主函数

多进程ffplay拉流(并保存视频)

import cv2
import multiprocessing as mp
# 这里需要依赖的还有 ffmpeg 
def image_save(q,url,camera_id):
    print(url)
    cap = cv2.VideoCapture(url)
    ret, frame = cap.read()
    print(ret) # 这里会返回是否正常返回流，正常返回True
    fourcc = cv2.VideoWriter_fourcc(*'XVID') # 创建本地文件
    fps = cap.get(cv2.CAP_PROP_FPS)
    size = (int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)),int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)))
    save_path='{}.avi'.format(camera_id)
    print(save_path)
    out = cv2.VideoWriter(save_path, fourcc, fps, size)
    # while ret:
    while ret:
        ret,frame=cap.read()
        if not ret:
            cap = cv2.VideoCapture(url)
            print('the url {} connect again'.format(url))
            ret, frame = cap.read()
        # print('{}'.format(camera_id),ret)
        try:
            cv2.imshow('{}'.format(url),frame)
        except (cv2.error) as e:
            print('except:', e)
            continue
        if cv2.waitKey(1)&0xFF==ord('q'):
            break

        out.write(frame)
    cv2.destroyAllWindows()
    cap.release()
def main(urls,camera_ids):
    print(urls)
    mp.set_start_method(method='spawn')  # init
    queues = [mp.Queue(maxsize=2) for _ in urls]
    processes = []
    for queue, url,camera_id in zip(queues, urls,camera_ids):
        processes.append(mp.Process(target=image_save, args=(queue, url,camera_id)))
        # processes.append(mp.Process(target=image_get, args=(queue, camera_ip)))
    for process in processes:
        process.daemon = True
        process.start()
    for process in processes:
        process.join()

if __name__ == '__main__':
    """begin detection cemara"""
    urls,camera_ids = [],[]
    for i in range(1, 5):# 1 2
        camera_ids.append(int(i))
        urls.append('rtmp://127.0.0.1:1935/live/{}'.format(i))
    # print(camera_ids)
    main(urls,camera_ids)  # 调用主函数

多进程和多线程的配合使用

import multiprocessing
import threading

def foo():
    print 'threading.current_thread(): ', threading.current_thread()

def bar():
    threads = []
    for _ in range(4): # each Process creates a number of new Threads
            thread = threading.Thread(target=foo)
            threads.append(thread)
    for thread in threads:
            thread.start()
            thread.join()

if __name__ == "__main__":
    processes = []
    for _ in range(3):
            p = multiprocessing.Process(target=bar) # create a new Process
            processes.append(p)
    for process in processes:
            process.start()
            process.join()