Python ThreadPoolExecutor监听停机信号实现优雅退出问题排查

核心错误1：ThreadPoolExecutor提交方式错误 #

你原来的写法executor.submit( producer(queue, engine_do) )会立即同步调用producer函数，主线程直接卡在producer的死循环中，消费者永远不会被启动。正确的submit用法是传入函数对象本身，再依次传递参数：

futures.append(executor.submit(producer, queue, running_flag))
futures.append(executor.submit(consumer, queue, running_flag))

核心错误2：运行状态变量没有同步 #

你使用全局布尔变量engine_do控制线程运行，但布尔是不可变类型，你将它作为参数传入线程函数时，传递的是值拷贝，后续全局变量修改为False后，线程内部的self.running还是True，所以收到信号后线程不会停止。
这里使用threading.Event作为状态控制标志，它是线程安全的可变对象，修改和读取都会同步到所有线程。

核心错误3：消费者代码存在类型异常 #

print("Read: " + len(items))中len(items)是整数，不能直接和字符串拼接，会抛出TypeError导致消费者线程直接崩溃退出，需要转为字符串。

核心错误4：优雅退出逻辑执行顺序错误 #

你用with语句包裹ThreadPoolExecutor，with块结束时会默认等待所有线程退出，所以不需要额外调用wait(futures)。同时原来的异常捕获位置不对，信号触发后需要先通知生产者停止，再等待队列消费完成后再停止消费者。

from pypac import PACSession # 继承系统代理设置
from queue import Queue
from requests.models import PreparedRequest
from concurrent.futures import ThreadPoolExecutor
import threading
import requests
import signal
import time    

class Producer:
    def __init__(self, running_event: threading.Event):
        self.url = "https://reqres.in/api/users"
        self.session = PACSession()
        self.running_event = running_event

    def put_into_queue(self, queue):
        page = 1
        request = PreparedRequest()

        # 用is_set()判断是否继续运行
        while self.running_event.is_set():
            request.prepare_url(self.url, {"page" : "1"})

            if not queue.full():                
                response = self.session.get(request.url)     

                if response.status_code == requests.codes.ok:                        
                    json_response = response.json()
                    response_length = len(json_response)
                    if response_length > 0 :
                        queue.put(json_response)
                        print(f"Inserted: {response_length}")  
                        # page += 1
                        time.sleep(10)


class Consumer:
    def __init__(self, running_event: threading.Event):
        self.running_event = running_event

    def read_from_queue(self, queue):
        # 只要运行标志没结束或者队列还有数据，就继续消费
        while self.running_event.is_set() or not queue.empty():
            try:
                # 超时1秒，避免一直阻塞无法检测运行标志
                items = queue.get(timeout=1)
                print(f"Read: {len(items)}")
                queue.task_done()
            except:
                # 队列为空超时，继续循环检测状态
                continue


def producer(queue, running_event):
    producer = Producer(running_event) 
    producer.put_into_queue(queue)

def consumer(queue, running_event):
    consumer = Consumer(running_event) 
    consumer.read_from_queue(queue)


def handler(signum, frame):
    print(f'收到信号，信号编号={signum}，开始优雅退出')
    global running_event
    # 先设置运行标志为False，通知生产者停止生产
    running_event.clear()


if __name__ == "__main__":  
    BUF_SIZE = 400
    THREAD_COUNT = 2
    queue = Queue(BUF_SIZE)

    # 用Event代替布尔变量做线程状态控制
    running_event = threading.Event()
    running_event.set() # 初始状态为运行中

    # 注册信号处理函数
    signal.signal(signal.SIGINT, handler)
    signal.signal(signal.SIGTERM, handler)
    # Windows特有的信号，非Windows系统可以注释掉
    try:
        signal.signal(signal.SIGBREAK, handler)
    except AttributeError:
        pass

    with ThreadPoolExecutor(THREAD_COUNT) as executor:
        # 正确提交任务：传函数对象，后面跟参数
        producer_future = executor.submit(producer, queue, running_event)
        consumer_future = executor.submit(consumer, queue, running_event)
    
    # 等待队列所有任务都被消费完成
    queue.join()
    print("所有任务处理完成，程序退出")

1. 收到CTRL+C或者kill信号后，首先设置running_event为停止状态，生产者线程检测到后会停止拉取新数据
2. 消费者线程会继续运行直到队列中所有剩余数据都被处理完成后才会退出
3. 避免了空轮询问题，CPU占用更低
4. 所有线程操作都是线程安全的，不会出现数据丢失

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