Python多线程场景下函数响应延迟时终止线程的实现方案

问题分析 #

• 你当前使用的threading.Thread没有提供外部直接终止运行中线程的接口，原有逻辑里的thread.join(10)仅会阻塞主线程等待指定线程10秒，不会主动终止超时线程，且仅在活跃线程数等于50时才会触发等待逻辑，完全无法实现每个线程独立的10秒超时控制。
• 原有并发控制逻辑存在误差：threading.active_count()统计的数量包含主线程，判断等于50的逻辑实际最多只能同时运行49个工作线程，不符合预期。

注意：原生Python线程不支持从外部强制杀死，所有单线程超时终止的逻辑都需要线程内部主动响应退出信号，否则只能将线程设为守护线程，等待主进程退出时统一销毁。

推荐实现方案（使用concurrent.futures.ThreadPoolExecutor） #

用Python标准库自带的线程池实现，自带超时控制能力，同时可以轻松控制并发数，代码更简洁易维护：

import threading
from concurrent.futures import ThreadPoolExecutor, TimeoutError

# 并发工作线程数，可根据需求调整
MAX_WORKERS = 49
# 超时时间，单位秒
TASK_TIMEOUT = 10

def getHLS(line, linenumber):
    # 此处保留你原有的getHLS业务逻辑即可
    pass

if __name__ == "__main__":
    # 读取链接文件，指定编码避免乱码
    with open("links.txt", "r", encoding="utf-8") as f:
        lines = f.readlines()
    
    # 初始化线程池
    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        future_mapping = {}
        # 批量提交所有任务
        for linenumber, line in enumerate(lines, start=1):
            future = executor.submit(getHLS, line.strip(), linenumber)
            future_mapping[future] = (line, linenumber)
        
        # 逐个检查任务运行状态，超时直接终止
        for future in future_mapping:
            try:
                # 等待指定时间获取结果，超时抛出异常
                result = future.result(timeout=TASK_TIMEOUT)
                # 此处可添加正常返回后的结果处理逻辑
            except TimeoutError:
                _, lineno = future_mapping[future]
                print(f"第{lineno}行任务运行超时，已终止")

原生Thread兼容方案（需修改getHLS逻辑） #

如果你一定要用原生threading.Thread实现，需要给每个线程新增退出标志位，在getHLS的业务逻辑中定期检查标志位，超时后主动退出：

import threading
import time

MAX_ACTIVE_WORKERS = 49
TASK_TIMEOUT = 10
thread_list = []

def getHLS(line, linenumber, exit_flag):
    # 原有业务逻辑中需要定期检查exit_flag状态，触发则主动退出
    # 示例：如果有循环逻辑，每轮循环都加如下判断
    # if exit_flag.is_set():
    #     return
    pass

if __name__ == "__main__":
    with open("links.txt", "r", encoding="utf-8") as f:
        lines = f.readlines()

    for linenumber, line in enumerate(lines, start=1):
        # 为每个线程生成独立的退出标志
        exit_flag = threading.Event()
        thread = threading.Thread(
            target=getHLS,
            args=(line.strip(), linenumber, exit_flag),
            daemon=False
        )
        # 存储线程、退出标志、启动时间、行号
        thread_list.append((thread, exit_flag, time.time(), linenumber))
        thread.start()
        
        # 达到并发上限时，清理已完成/超时的线程
        while threading.active_count() > MAX_ACTIVE_WORKERS + 1:
            current_ts = time.time()
            # 倒序遍历避免删除元素导致的索引错误
            for i in range(len(thread_list)-1, -1, -1):
                t, flag, start_ts, lineno = thread_list[i]
                if not t.is_alive():
                    t.join()
                    thread_list.pop(i)
                elif current_ts - start_ts > TASK_TIMEOUT:
                    flag.set()
                    t.join()
                    thread_list.pop(i)
                    print(f"第{lineno}行任务运行超时，已终止")
            time.sleep(0.1)

    # 最后清理所有剩余线程
    current_ts = time.time()
    for t, flag, start_ts, lineno in thread_list:
        if t.is_alive():
            if current_ts - start_ts > TASK_TIMEOUT:
                flag.set()
            t.join()

内容的提问来源于stack exchange，提问作者CDoc