# -*- coding: utf-8 -*- from queue import Queue from threading import Thread from time import time, sleep from traceback import format_exc from loguru import logger from enums.ExceptionEnum import ExceptionType from enums.RecordingStatusEnum import RecordingStatus from exception.CustomerException import ServiceException from util.Cv2Utils import check_video_stream, clear_pull_p, build_video_info2, pull_read_video_stream2 from util.QueUtil import put_queue, get_no_block_queue, clear_queue, put_queue_result class PullStreamThread(Thread): __slots__ = ('_command', '_pull_queue', '_hb_queue', '_fb_queue', '_msg', '_context') def __init__(self, *args): super().__init__() self._msg, self._context, self._pull_queue, self._hb_queue, self._fb_queue, self._frame_num = args self._command = Queue() def sendEvent(self, result): put_queue(self._command, result, timeout=10, is_ex=False) class RecordingPullStreamThread(PullStreamThread): def run(self): msg, context, frame_num = self._msg, self._context, self._frame_num request_id, pull_url = msg["request_id"], msg['pull_url'] service = context["service"] pull_stream_timeout = int(service["recording_pull_stream_timeout"]) read_stream_timeout = int(service["cv2_read_stream_timeout"]) service_timeout = int(service["timeout"]) command_queue, pull_queue, fb_queue, hb_queue = self._command, self._pull_queue, self._fb_queue, self._hb_queue width, height, width_height_3, all_frames, w, h = None, None, None, 0, None, None read_start_time, pull_p, ex = None, None, None frame_list, frame_index_list = [], [] stop_ex = True pull_stream_start_time = time() try: logger.info("录屏拉流线程开始启动, requestId: {}", request_id) cv2_init_num, init_pull_num, concurrent_frame = 0, 1, 1 start_time = time() while True: # 检查任务是否超时 if time() - start_time > service_timeout: logger.error("录屏拉流超时, requestId: {}", request_id) raise ServiceException(ExceptionType.TASK_EXCUTE_TIMEOUT.value[0], ExceptionType.TASK_EXCUTE_TIMEOUT.value[1]) # 最终停止拉流 event = get_no_block_queue(command_queue) if event is not None: # 当接收到停止指令，说明不会再处理视频帧了, 直接退出 if 'stop' == event.get("command"): if len(frame_list) > 0: put_queue(pull_queue, (4, (frame_list, frame_index_list, all_frames)), timeout=1) logger.info("录屏拉流线程开始停止, requestId: {}", request_id) break # 主进程异常，停止子线程 if 'stop_ex' == event.get("command"): logger.info("录屏异常拉开始停止拉流线程, requestId: {}", request_id) stop_ex = False break # 如果是离线拉流 if pull_url.startswith('http'): if check_video_stream(width, height): logger.info("开始重新获取视频信息: {}次, requestId: {}", cv2_init_num, request_id) # 当是离线地址重试3次还是拉取不到视频流，关闭拉流管道，返回失败信息 # 目前改为等待5分钟 # if cv2_init_num > 3: if time() - start_time > 300: logger.info("离线拉流重试失败, 重试次数: {}, requestId: {}", cv2_init_num, request_id) raise ServiceException(ExceptionType.OR_VIDEO_ADDRESS_EXCEPTION.value[0], ExceptionType.OR_VIDEO_ADDRESS_EXCEPTION.value[1]) cv2_init_num += 1 width, height, width_height_3, all_frames, w, h = build_video_info2(pull_url, request_id) if width is not None: put_queue(hb_queue, {"status": RecordingStatus.RECORDING_RUNNING.value[0]}, timeout=2) else: # if cv2_init_num < 2: if time() - start_time < 300: put_queue(hb_queue, {"status": RecordingStatus.RECORDING_RETRYING.value[0]}, timeout=2) continue # 当离线视频时, 队列满了, 等待1秒后再试 if pull_queue.full(): logger.info("pull拉流队列满了: {}, requestId: {}", pull_queue.qsize(), request_id) sleep(1) continue # 如果是实时拉流 else: if check_video_stream(width, height): logger.info("开始重新获取视频信息: {}次, requestId: {}", cv2_init_num, request_id) pull_stream_init_timeout = time() - pull_stream_start_time if len(frame_list) > 0: put_queue(pull_queue, (4, (frame_list, frame_index_list, all_frames)), timeout=1) frame_list, frame_index_list = [], [] if pull_stream_init_timeout > pull_stream_timeout: logger.error("开始拉流超时, 超时时间：{}, requestId:{}", pull_stream_init_timeout, request_id) raise ServiceException(ExceptionType.PULLSTREAM_TIMEOUT_EXCEPTION.value[0], ExceptionType.PULLSTREAM_TIMEOUT_EXCEPTION.value[1]) cv2_init_num += 1 width, height, width_height_3, all_frames, w, h = build_video_info2(pull_url, request_id) if width is not None: put_queue(hb_queue, {"status": RecordingStatus.RECORDING_RUNNING.value[0]}, timeout=1) else: if cv2_init_num < 3: put_queue(hb_queue, {"status": RecordingStatus.RECORDING_RETRYING.value[0]}, timeout=1) sleep(1) continue pull_stream_start_time = time() cv2_init_num = 1 frame, pull_p, width, height = pull_read_video_stream2(pull_p, pull_url, width, height, width_height_3, w, h, request_id) if frame is None: if pull_url.startswith('http'): clear_pull_p(pull_p, request_id) logger.info("总帧数: {}, 当前帧数: {}, requestId: {}", all_frames, concurrent_frame, request_id) if len(frame_list) > 0: put_queue(pull_queue, (4, (frame_list, frame_index_list, all_frames)), timeout=1) if concurrent_frame < all_frames - 100: logger.info("离线拉流异常结束：requestId: {}", request_id) raise ServiceException(ExceptionType.READSTREAM_TIMEOUT_EXCEPTION.value[0], ExceptionType.READSTREAM_TIMEOUT_EXCEPTION.value[1]) logger.info("离线拉流线程结束, requestId: {}", request_id) break else: logger.info("获取帧为空, 开始重试: {}次, requestId: {}", init_pull_num, request_id) if len(frame_list) > 0: put_queue(pull_queue, (4, (frame_list, frame_index_list, all_frames)), timeout=1) frame_list, frame_index_list = [], [] if read_start_time is None: read_start_time = time() pull_stream_read_timeout = time() - read_start_time if pull_stream_read_timeout > read_stream_timeout: logger.info("拉流过程中断了重试超时, 超时时间: {}, requestId: {}", pull_stream_read_timeout, request_id) raise ServiceException(ExceptionType.READSTREAM_TIMEOUT_EXCEPTION.value[0], ExceptionType.READSTREAM_TIMEOUT_EXCEPTION.value[1]) init_pull_num += 1 continue init_pull_num = 1 read_start_time = None if pull_queue.full(): sleep(1) logger.info("pull拉流队列满了:{}, requestId: {}", pull_queue.qsize(), request_id) continue frame_list.append(frame) frame_index_list.append(concurrent_frame) if len(frame_list) >= frame_num: put_queue(pull_queue, (4, (frame_list, frame_index_list, all_frames)), timeout=1) frame_list, frame_index_list = [], [] concurrent_frame += 1 del frame except ServiceException as s: ex = s.code, s.msg except Exception: logger.exception("实时拉流异常: {}, requestId:{}", format_exc(), request_id) ex = ExceptionType.SERVICE_INNER_EXCEPTION.value[0], ExceptionType.SERVICE_INNER_EXCEPTION.value[1] finally: clear_pull_p(pull_p, request_id) if stop_ex: if ex: error_code, error_msg = ex result = put_queue_result(pull_queue, (1, error_code, error_msg), timeout=3) else: result = put_queue_result(pull_queue, (2,), timeout=3) if result: # 3分钟超时时间 cr_time = time() while time() - cr_time < 180: event = get_no_block_queue(command_queue) if event is not None: # 当接收到停止指令，说明不会再处理视频帧了, 直接退出 if 'stop' == event.get("command"): logger.info("录屏拉流线程开始停止, requestId: {}", request_id) break sleep(1) clear_queue(command_queue) clear_queue(pull_queue) clear_queue(hb_queue) del frame_list, frame_index_list logger.info("录屏拉流线程结束, requestId: {}", request_id)