如何在单独的文件中访问类外部定义的变量,而不会遇到循环导入?

How do I access a variable defined outside my class in a separate file without running into circular imports?

提问人:Vinayak 提问时间:7/6/2023 更新时间:7/7/2023 访问量:51

问:

我想访问在类的实例中定义的。这是我的代码my_variablemain.pySecondThread

main.py

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}")

    first_thread = threading.Thread(target=FirstThread().run, daemon=True, name='FirstThread')
    first_thread.start()

    second_thread = threading.Thread(target=SecondThread().run, daemon=True, name='SecondThread')
    second_thread.start()

    while True:
        first_thread.join(1)

if __name__ == "__main__":
    start()

first_thread.py

import threading
import time

from second_thread import SecondThread

class FirstThread:
    def __init__(self):
        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...")
                    second_thread = threading.Thread(target=SecondThread().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...")

second_thread.py

import threading
import time

from main import my_variable

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

    def run(self):
        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...")

如果我运行,我会得到:python3 main.py

ImportError: cannot import name 'SecondThread' from partially initialized module 'second_thread' (most likely due to a circular import) (/home/ubuntu/test/second_thread.py)

如果我删除 并在 in second_thread.py之前添加一个右键,我会得到:from main import my_variableglobal my_variableif not my_variable.is_set():

NameError: name 'my_variable' is not defined

如果我在单个 main.py 文件中定义所有线程类,而不是将它们放在单独的文件中,这将起作用,但我不想这样做。

我怎样才能避免这种情况?

python-3.x 变量 作用域 python-multithreading

评论

1赞 yedpodtrzitko 7/6/2023
尝试将变量作为参数传递给每个线程,而不是将其导入到每个线程。

答:

1赞 blhsing 7/6/2023 #1

通常,您可以通过将导入会导致循环导入问题的模块的语句放在函数中来避免循环导入,以便在有问题的模块导入当前模块时不会执行该语句。import

然而,在这种情况下,当主程序运行时,它是由解释器以名称而不是 “导入”的,所以当 时,导入系统找不到导入缓存中指定的模块,因此继续再次导入主模块,创建另一个对象,该对象不与主程序运行时创建的对象共享。__main__mainsecond_thread.pyimport mainmainEvent

因此,要访问解释器最初加载的主模块,您需要指定名称:__main__

second_thread.py

import threading
import time
import sys    

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...")

此外,你不必要地在打电话后打电话,所以又变成了。删除调用,程序将按预期工作。my_variable.clear()my_variable.set()my_variable.is_set()False

演示:https://replit.com/@blhsing/BitterAdmirableStructs