Python多线程中如何通过输入控制同一类实例的方法执行？

先修复原代码的语法问题，再通过线程安全的机制实现实例共享与状态控制：

原代码核心问题 #

• Hello类构造方法写错为__init(self)，正确应为__init__(self)
• say_hello方法中if should_say_bye = True:是赋值语句，而非判断逻辑，且全局变量跨线程访问存在安全隐患
• launch函数内的Hello实例是局部变量，input_reader无法直接访问
• 全局变量should_say_bye未在Hello类中声明，无法直接调用

修改后的代码 #

main.py #

import threading
import time
from app import Hello

# 线程安全的退出信号，全局存储Hello实例
exit_signal = threading.Event()
hello_instance = None

def launch(thread_name):
    global hello_instance
    hello_instance = Hello(exit_signal)
    hello_instance.say_hello()

def input_reader(thread_name):
    x = input()
    if x.lower() == "q":
        exit_signal.set()  # 触发退出信号
        print("程序即将退出。")

if __name__ == "__main__":
    try:
        # 用threading.Thread替代底层的_thread模块，更易维护
        threading.Thread(target=launch, args=("App",)).start()
        threading.Thread(target=input_reader, args=("Input_Reader",)).start()
    except Exception as e:
        print(f"Error: {e}")

    # 主线程监听退出信号，触发后结束程序
    while not exit_signal.is_set():
        time.sleep(0.1)

app.py #

import time

class Hello:
    def __init__(self, exit_signal):
        self.exit_signal = exit_signal

    def say_hello(self):
        i = 0
        # 循环执行直到收到退出信号
        while not self.exit_signal.is_set():
            print(f"Hello {i}")
            i += 1
            time.sleep(2)
        # 收到信号后调用say_bye方法
        self.say_bye()

    def say_bye(self):
        print("Bye")

方案说明 #

1. 线程安全信号传递：用threading.Event替代全局变量，它是专门用于线程间信号通知的线程安全对象，避免了全局变量的竞态问题
2. 共享Hello实例：将Hello实例存储为全局变量，input_reader通过触发exit_signal间接控制实例的行为，确保使用的是launch线程中创建的同一个实例
3. 修复语法错误：修正了构造方法和条件判断的语法问题
4. 合理线程管理：用threading.Thread替代_thread模块，功能更完善；主线程通过监听信号结束，避免无意义的死循环

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