Python主线程超时后单工作线程被终止但多线程存活的原因及解决方案咨询

嘿，这个问题挺有意思的！我来帮你拆解一下问题根源，再给你几个靠谱的解决方案。

首先，咱们先搞清楚为啥会出现这种“单线程死、多线程活”的诡异区别——核心原因大概率是你用的@timeout装饰器依赖了Unix系统的SIGALRM信号，而这个信号的投递逻辑在单/多线程场景下有差异。

问题根源：SIGALRM信号的投递随机性与线程中断 #

在macOS这类Unix-like系统中，绝大多数超时装饰器是靠发送SIGALRM信号实现的：到了超时时间，装饰器给进程发一个SIGALRM，触发主线程抛出超时异常。但这里有个关键细节：SIGALRM是进程级信号，内核会把它发给进程中任意一个没有阻塞该信号的线程。

咱们分两种场景看：

1. 单工作线程场景：
   主线程在th.join()处阻塞等待，工作线程在time.sleep(120)处也阻塞。超时后内核要选一个线程发信号，这时候系统可能会选中工作线程（因为它是除了主线程之外唯一的活跃线程）。
   而sleep()是可被信号中断的系统调用，工作线程收到SIGALRM后会直接抛出InterruptedError。如果你的_execute函数没捕获这个异常，线程就会直接终止，看起来像是被“杀死”了。

2. 双工作线程场景：
   主线程同样在join时阻塞，但此时系统有两个工作线程+一个主线程，内核更倾向于把SIGALRM发给主线程（毕竟它是发起超时逻辑的“源头”线程）。主线程抛出超时异常后退出，而两个工作线程完全没收到信号，所以能安安稳稳睡够120秒。

你可以做个小实验验证：在_execute函数里加个异常捕获，看看单线程的工作线程是不是触发了InterruptedError：

def _execute(self, connector_init_params, code, task, ipparam, order, output_queue):
    try:
        start_time = time.time()
        time.sleep(120)
        print(f"Worker {order} finished sleep")
    except InterruptedError:
        print(f"Worker {order} got interrupted by SIGALRM!")
        # 可以尝试补完剩余的sleep时间
        remaining = 120 - (time.time() - start_time)
        if remaining > 0:
            time.sleep(remaining)
            print(f"Worker {order} finished remaining sleep")

如果单线程的worker打印了got interrupted，那咱们的判断就坐实了。

解决方案：让工作线程免疫SIGALRM信号 #

既然问题出在SIGALRM误杀了单工作线程，那咱们的核心思路就是让工作线程对这个信号“免疫”。这里给你两个最实用的方案：

方案1：在工作线程中阻塞SIGALRM

Python的signal模块支持给单个线程设置信号掩码，咱们可以在工作线程的入口函数开头，把SIGALRM信号给阻塞掉——这样这个线程就再也收不到SIGALRM了，sleep也不会被打断。

修改你的_execute函数：

import signal
import time

def _execute(self, connector_init_params, code, task, ipparam, order, output_queue):
    # 给当前线程设置信号掩码，阻塞SIGALRM
    signal.pthread_sigmask(signal.SIG_BLOCK, {signal.SIGALRM})
    
    try:
        # 你的原有逻辑：包括exec()和sleep
        time.sleep(120)
        print(f"Worker {order} finished successfully")
        # 把结果写入output_queue（如果需要的话）
    finally:
        # 可选：线程结束前解除阻塞，不过不解除也没啥问题，线程退出后掩码自动失效
        # signal.pthread_sigmask(signal.SIG_UNBLOCK, {signal.SIGALRM})
        pass

这个方案最直接，完全不影响主线程的超时逻辑，工作线程该干嘛干嘛。

方案2：替换基于SIGALRM的超时装饰器

如果不想碰信号掩码，你可以换一种不依赖SIGALRM的超时实现——比如用监控线程来触发超时，而不是发信号。

比如自己实现一个简单的超时装饰器：

import threading
import time
from functools import wraps

def timeout(seconds):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            result_container = []
            exc_container = []
            
            def target():
                try:
                    # 把原函数的执行放到一个子线程里
                    result = func(*args, **kwargs)
                    result_container.append(result)
                except Exception as e:
                    exc_container.append((type(e), e, e.__traceback__))
            
            # 启动子线程执行原函数
            worker_thread = threading.Thread(target=target)
            worker_thread.start()
            # 等待指定时间，超时就返回
            worker_thread.join(seconds)
            
            if worker_thread.is_alive():
                # 原函数还在跑，抛出超时异常
                raise TimeoutError(f"Function timed out after {seconds}s")
            # 如果原函数抛出了异常，重新抛出
            if exc_container:
                raise exc_container[0][0](exc_container[0][1]).with_traceback(exc_container[0][2])
            # 返回原函数的结果
            return result_container[0] if result_container else None
        return wrapper
    return decorator

这个装饰器的原理是把你的_execute_parallel函数放到一个子线程里执行，超时后直接抛出异常，全程不发任何进程级信号。这样不管你开多少个工作线程，都不会被信号打断，能安安稳稳执行到完成。

不过这个方案有个小注意点：原函数的执行上下文会从主线程变成装饰器的子线程，如果你的代码依赖主线程的某些状态（比如线程局部变量、GUI主线程等），可能需要调整。

最后确认 #

不管用哪个方案，记得确保你的工作线程确实不是守护线程（你已经做到了，默认threading.Thread的daemon参数是False），这样即使主线程退出，进程也会等所有非守护线程完成后才终止。

内容来源于stack exchange