Python守护线程：如何通过子线程触发标志让主线程退出？

你遇到的核心问题是input()是阻塞式调用——主线程卡在input('Enter input: ')的时候，完全不会执行后续的循环条件判断，哪怕子线程已经把self.done设为True，主线程也得等用户输入完才会察觉到，自然没法及时退出。

下面给你两种可行的解决思路：

方案1：用select实现带超时的输入（Unix/Linux适用） #

利用select模块让输入操作每隔一段时间超时一次，这样主线程就能定期检查self.done状态，不用一直卡在输入等待上。

修改后的代码：

import threading
import time
import sys
import select

class Test:
    def __init__(self):
        self.done = False
        self.inp = ''
        
    def worker(self):
        count = 0
        while True:
            if self.inp == 'start':
                print(f"Hello from Worker {count}")
                count += 1
                time.sleep(1)
                if count == 3:
                    self.done = True
                    return
                
    def start(self):
        while not self.done:
            # 检查是否有输入，超时1秒
            ready, _, _ = select.select([sys.stdin], [], [], 1)
            if ready:
                self.inp = sys.stdin.readline().strip()
                print(self.done)
            # 每次超时后都会回到循环开头检查self.done
            time.sleep(0.1)
        
        print("Main Thread ending")
        sys.exit()

test = Test()
thread = threading.Thread(target=test.worker)
thread.daemon = True
thread.start()
test.start()

方案2：把输入逻辑放到独立线程（跨平台通用） #

让主线程专注于检查self.done状态，输入操作交给另一个子线程处理，这样主线程不会被阻塞，能及时响应退出信号。

修改后的代码：

import threading
import time
import sys

class Test:
    def __init__(self):
        self.done = False
        self.inp = ''
        
    def worker(self):
        count = 0
        while True:
            if self.inp == 'start':
                print(f"Hello from Worker {count}")
                count += 1
                time.sleep(1)
                if count == 3:
                    self.done = True
                    return
                
    def input_handler(self):
        while not self.done:
            self.inp = input('Enter input: ')
            print(self.done)
            time.sleep(1)
        
    def start(self):
        # 启动输入线程
        input_thread = threading.Thread(target=self.input_handler)
        input_thread.daemon = True
        input_thread.start()
        
        # 主线程循环检查done状态
        while not self.done:
            time.sleep(0.5)
        
        print("Main Thread ending")
        sys.exit()

test = Test()
thread = threading.Thread(target=test.worker)
thread.daemon = True
thread.start()
test.start()

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