导入一个 python class 的非打包模块,并在它自己的线程中导入 运行

Import a python class of non packaged module, and run in it's own thread

详细介绍了如何使用 importlib 导入模块,并检查以按名称查找包含的 class。

虽然这在使用 spec.loader.exec_module(module) 执行模块时有效,但是 'MyClass' 在 module 中找不到 运行 线程中完全相同的代码,就像这样: threading.Thread(target=spec.loader.exec_module, args=[module])

我已经建立了一个工作示例,可以运行看到这个效果以简单的方式演示,只需将以下内容复制到两个文件中并执行第一个。

文件 1:

# test1.py

import importlib.util
import inspect
import threading

my_file = 'thread_test'
path = "thread_test.py"

spec = importlib.util.spec_from_file_location(my_file, path)
module = importlib.util.module_from_spec(spec)

# Comment out one of the following snippets to test each version:

# OPTION 1:
# Non-Threaded Version, Works:
spec.loader.exec_module(module)

# OPTION 2:
# Threaded Version, Doesn't work (Causes cls to not be found in the following loop):
# thread = threading.Thread(target=spec.loader.exec_module, args=[module])
# thread.start()

cls = None
for member in inspect.getmembers(module):
    if member[0] == 'MyClass':
        cls = member[1]

cls.color = 'blue'

cls().PrintColor()

文件 2:

# thread_test.py

class MyClass:
    color = None
    
    def PrintColor(self):
        print(self.color)

为什么模块在其自己的线程中执行时不包含 MyClass,但它随后在同一线程中执行?

您只启动了线程,但您必须加入它并等待它有 运行:thread.join()。否则模块将无法正确导入。

完整示例:

import importlib.util
import inspect
import threading

my_file = 'thread_test'
path = "thread_test.py"

spec = importlib.util.spec_from_file_location(my_file, path)
module = importlib.util.module_from_spec(spec)

# Comment out one of the following snippets to test each version:

# OPTION 1:
# Non-Threaded Version, Works:
# spec.loader.exec_module(module)

# OPTION 2:
# Threaded Version, Doesn't work (Causes cls to not be found in the following loop):
thread = threading.Thread(target=spec.loader.exec_module, args=[module])
thread.start()
thread.join()

cls = None
for member in inspect.getmembers(module):
    if member[0] == 'MyClass':
        cls = member[1]

cls.color = 'blue'

cls().PrintColor()

结果:

blue

经过与其他人的讨论,我得出了以下解决方案。 问题可以通过 运行 加载模块和 class 在它自己的线程中 来解决。本质上只是更快地启动线程。

所以我们把它全部放到下面的函数中:

# Run this in it's own thread
def my_function(module, class_name):
    spec.loader.exec_module(module)
    
    # Get class
    for member in inspect.getmembers(module):
        if member[0] == class_name:
            my_class = member[1]
        if my_class is None:
            my_print("Error: Could not find class " + class_name + '!')
            return

    # Set properties
    my_class.color = 'blue'
    
    # Execute code
    cls().PrintColor()

导入:

spec = importlib.util.spec_from_file_location(file, path)
module = importlib.util.module_from_spec(spec)

然后简单地在它自己的线程中执行这个函数:

# Start thread
thread = threading.Thread(target=my_function, args=[module, class_name])
thread.start()