google_ai_studio/job_video_processing.py

import time
from concurrent.futures import ThreadPoolExecutor
from video_processing.video_processing import VideoProcessing

max_workers = 5  # 最大线程数


def video_ai_task_start():
    redis_task_list = ['task:video_ai_top', 'task:video_ai_top', 'task:video_ai_recommend', 'task:video_ai_recommend', 'task:video_ai_recommend']
    futures = []  # 用来存储任务的 Future 对象
    task_index = 0

    with ThreadPoolExecutor( max_workers=max_workers ) as executor:
        while True:
            futures = [f for f in futures if not f.done()]

            if len(futures) < max_workers:
                try:
                    # 提交新任务
                    future = executor.submit( process_video_ai, redis_task_list[task_index] )
                    futures.append( future )

                    task_index += 1
                    if task_index >= len( redis_task_list ):
                        task_index = 0

                    time.sleep( 1 )

                except Exception as e:
                    print( f"异常信息: {e}" )
                    time.sleep(1)
            else:
                time.sleep(1)


def process_video_ai(redis_task):
    try:
        print( f"开始执行任务: {redis_task}" )
        video_processor = VideoProcessing()
        video_processor.get_video( redis_task )
        print( f"执行完成: {redis_task}" )
        time.sleep( 5 )  # 模拟处理时间
    except Exception as e:
        print( f"处理任务时出现异常: {e}" )
        time.sleep( 5 )  # 出现异常时，等待5秒再处理


if __name__ == '__main__':
    video_ai_task_start()