Python多线程间如何共享数据及实现线程启停、变量实时读取

• 全局变量未提前初始化：你没有在全局作用域预先定义countval变量，只有当count函数执行到for循环给变量赋值后，这个全局变量才会被创建，所以print_stuff函数在计数逻辑未执行时访问countval就会抛出变量不存在的错误。
• 线程实例启动限制：Python的threading.Thread实例仅允许调用一次start()方法，重复调用会直接抛出RuntimeError，所以无法通过重复调用同一个线程实例的start()方法实现按需重启的需求。

推荐将计数器逻辑封装为独立类，使用threading.Event实现按需触发计数，避免全局变量的混乱和线程重复创建的问题，同时支持重置、实时读值的需求，实现代码如下：

import threading
import time

class ReusableCounter:
    def __init__(self, count_limit=3):
        self.countval = 0
        self.count_limit = count_limit
        # 计数触发信号，用于控制计数启动
        self.start_signal = threading.Event()
        # 停止信号，用于退出线程
        self.stop_signal = threading.Event()
        # 启动后台计数线程，线程常驻等待触发
        self.thread = threading.Thread(target=self._count_loop, daemon=True)
        self.thread.start()
    
    def _count_loop(self):
        # 后台循环，等待触发信号后执行计数
        while not self.stop_signal.is_set():
            # 等待启动信号，无信号时阻塞不占CPU
            self.start_signal.wait()
            # 计数逻辑
            for self.countval in range(self.count_limit):
                print('looping')
                time.sleep(1)
            # 计数完成后重置触发信号，等待下一次启动
            self.start_signal.clear()
    
    def start_count(self):
        """触发一次计数"""
        if not self.start_signal.is_set():
            self.countval = 0
            self.start_signal.set()
    
    def reset_count(self):
        """重置计数器为0，可在计数运行中调用"""
        self.countval = 0
    
    def get_current(self):
        """获取当前计数值"""
        return self.countval
    
    def stop(self):
        """停止后台线程，程序退出前调用"""
        self.stop_signal.set()
        self.start_signal.set()
        self.thread.join()

# 业务逻辑示例
def print_stuff(counter: ReusableCounter):
    while True:
        input("按回车键启动计数")
        counter.start_count()
        while True:
            current = counter.get_current()
            if current == 2: # 对应range(3)的最大值2，计数完成后是2
                print("time out")
                break
            time.sleep(0.1)

if __name__ == "__main__":
    counter = ReusableCounter()
    print_thread = threading.Thread(target=print_stuff, args=(counter,))
    print_thread.start()
    # 可以在任意位置调用counter方法实现控制：
    # counter.reset_count() 重置计数
    # counter.get_current() 读当前值

• 线程安全：所有状态都封装在类实例中，避免全局变量的污染和访问冲突
• 按需触发：计数线程后台常驻，平时处于阻塞等待状态不消耗CPU，需要计数时调用start_count()即可触发，不需要重复创建线程实例
• 功能完备：支持运行中重置计数、随时读取计数值、计数完成自动等待下一次触发，完全匹配你的需求

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