多线程
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()
