Python 的 threading
模块有两个接口(interface)。在其中一个实例化 threading.Thread
并将您想要运行的函数传递给它,示意性地:
class MyClass:
def __init__(self):
self.my_vars = {}
def my_func(self):
# does stuff
self.my_vars = # set result of running the function here
mc = MyClass()
t = threading.Thread(target=mc.my_func) # also, pass arguments
# set some options to t, like t.setDaemon(True)
t.start()
在另一个你的子类中threading.Thread
并子类 run
方法,示意性:
class MyClass(threading.Thread):
def __init__(self):
super(MyClass,self).__init__(*args, **kwargs)
self.my_vars = {}
def run(self):
# does stuff
self.my_vars = # set result of running the function here
t = MyThreadedClass()
t.start()
我开始使用第一个,但在某些时候意识到每次我想启动一个线程来运行 my_func
时我都必须编写大量样板文件。 :我一直不得不提醒自己将参数传递给 my_func
的语法。 ,加上我必须编写多行来设置线程选项等。所以我决定转向第二种样式。这样我只需实例化我的类,然后调用 .start()
。请注意,现阶段的区别仅在于使用这些东西的容易程度,如 my_func
和run
完全一样。
但现在我意识到这使我的代码更难测试。之前如果我想测试my_func
在如此这般的争论下,我只需要 import
定义它的文件,然后在某些输入上运行它。或者我什至可以从 jupyter 笔记本上执行此操作并播放其输入以查看其输出。但使用第二种风格时,每次我想做一些像运行 my_func
这样简单的事情时,它带有一根线。
所以问题:有没有一种方法可以组织我的代码,使应用程序在自己的线程上运行它更干净,但当我想从笔记本中调用它时,它没有线程?
最佳答案
使类的对象可调用:
from abc import ABCMeta, abstractmethod
import threading
class Callable:
__metaclass__ = ABCMeta
@abstractmethod
def __call__(self): raise NotImplementedError
class MyCallable(Callable):
def __init__(self, x):
self.x = x
def __call__(self):
print('x=', self.x)
# without a thread:
callable = MyCallable(7)
callable()
# on a thread:
callableThread = threading.Thread(target=callable)
callableThread.start()
callableThread.join()
您可以不使用抽象基类和抽象方法,只需确保您的类定义 __call__
完成您当前需要做的工作的方法正在由您的 my_func
完成.
关于Python:我应该子类化 threading.Thread 吗?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/59520021/