如何避免循环导入并在独立文件的类中访问main.py变量？

你遇到的核心问题是循环导入：main.py导入SecondThread，而second_thread.py又尝试导入main.py中的my_variable，导致模块加载时互相依赖，出现部分初始化错误。下面提供两种可行的解决方法，优先推荐低耦合的参数传递方案。

方案一：通过参数传递变量（推荐） #

这种方式彻底消除循环导入，同时降低代码耦合度，更易维护。

修改second_thread.py #

移除对main的导入，通过构造函数接收my_variable并保存为实例变量：

import threading
import time

class SecondThread:
    def __init__(self, my_variable):
        self.my_variable = my_variable
        print(f"SecondThread[{threading.get_native_id()}]: Started")

    def run(self):
        for x in range(100):
            if not self.my_variable.is_set():
                print(f"{x}/100 SecondThread[{threading.get_native_id()}]: Sleeping for 10 seconds")
                time.sleep(10)
        print(f"SecondThread[{threading.get_native_id()}]: Exiting...")

修改first_thread.py #

同样通过构造函数接收my_variable，在重启SecondThread时传递该变量，并将SecondThread的导入移到模块末尾，避免加载时循环：

import threading
import time

class FirstThread:
    def __init__(self, my_variable):
        self.my_variable = my_variable
        print(f"FirstThread[{threading.get_native_id()}]: Started")

    def run(self):
        for x in range(100):
            if 'SecondThread' in [t.name for t in threading.enumerate()]:
                print(f"{x}/100 FirstThread[{threading.get_native_id()}]: Sleeping for 10 seconds")
            else:
                print(f"{x}/100 FirstThread[{threading.get_native_id()}]: SecondThread is not running yet...")
                print(f"{x}/100 FirstThread[{threading.get_native_id()}]: Waiting 5 seconds and checking again...")
                time.sleep(5)
                if not 'SecondThread' in [t.name for t in threading.enumerate()]:
                    print(f"{x}/100 FirstThread[{threading.get_native_id()}]: Starting SecondThread again...")
                    # 传递my_variable给新的SecondThread实例
                    second_thread = threading.Thread(target=SecondThread(self.my_variable).run, daemon=True, name='SecondThread')
                    second_thread.start()
                else:
                    print(f"{x}/100 FirstThread[{threading.get_native_id()}]: SecondThread instance was found!")

            time.sleep(10)
        print(f"FirstThread[{threading.get_native_id()}]: Exiting...")

# 延迟导入SecondThread，避免模块加载时的循环依赖
from second_thread import SecondThread

修改main.py #

创建线程实例时，将my_variable作为参数传递进去：

import threading
import time
import signal
import traceback

from second_thread import SecondThread
from first_thread import FirstThread

my_variable = threading.Event()

def stop_handler(signum, frame):
    print(f"Signal received: {signum}")
    print(f"Stack frame:\n{traceback.print_stack(frame)}")
    global my_variable
    print("Setting my_variable...")
    my_variable.set()
    print(f"my_variable[{threading.get_native_id()}]:: {my_variable}")
    time.sleep(5)
    print("Unsetting my_variable...")
    my_variable.clear()
    print(f"my_variable[{threading.get_native_id()}]:: {my_variable}")

def start():
    signal.signal(signal.SIGTERM, stop_handler)
    signal.signal(signal.SIGINT, stop_handler)

    print(f"my_variable[{threading.get_native_id()}]: {my_variable}")

    # 传递my_variable给FirstThread
    first_thread = threading.Thread(target=FirstThread(my_variable).run, daemon=True, name='FirstThread')
    first_thread.start()

    # 传递my_variable给SecondThread
    second_thread = threading.Thread(target=SecondThread(my_variable).run, daemon=True, name='SecondThread')
    second_thread.start()

    while True:
        first_thread.join(1)

if __name__ == "__main__":
    start()

方案二：延迟导入（快速临时解决） #

如果不想修改太多代码，可以在SecondThread的run方法内部延迟导入my_variable，避免模块级的循环导入：

import threading
import time

class SecondThread:
    def __init__(self):
        print(f"SecondThread[{threading.get_native_id()}]: Started")

    def run(self):
        # 仅在运行时导入，避开模块加载阶段的循环依赖
        from main import my_variable
        for x in range(100):
            if not my_variable.is_set():
                print(f"{x}/100 SecondThread[{threading.get_native_id()}]: Sleeping for 10 seconds")
                time.sleep(10)
        print(f"SecondThread[{threading.get_native_id()}]: Exiting...")

这种方式无需修改其他文件，但代码耦合度较高，SecondThread会依赖main.py中变量的名称，仅适合临时场景。

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