<p><em>注意</em>:不知何故,我忽略了这样一个事实,即您的<code>Thread</code>子类本身也是一个上下文管理器,因此下面的代码不会做出这种假设。然而,当使用更“通用”类型的线程时(使用<code>contextlib.ExitStack</code>之类的线程将不是一个选项),它可能会有所帮助。你知道吗</p>
<p>你的问题是有点轻的细节,所以我编了一些,但这可能是接近你想要的。它定义了一个<code>AsyncTaskListContextManager</code>类,该类具有支持上下文管理器协议所需的<code>__enter__()</code>和<code>__exit__()</code>方法(以及相关的<code>with</code>语句)。你知道吗</p>
<pre><code>import threading
from time import sleep
class SomeAsyncTask(threading.Thread):
def __init__(self, name, *args, **kwargs):
super().__init__(*args, **kwargs)
self.name = name
self.status_lock = threading.Lock()
self.running = False
def run(self):
with self.status_lock:
self.running = True
while True:
with self.status_lock:
if not self.running:
break
print('task {!r} running'.format(self.name))
sleep(.1)
print('task {!r} stopped'.format(self.name))
def stop(self):
with self.status_lock:
self.running = False
class AsyncTaskListContextManager:
def __init__(self, params_list):
self.threads = [SomeAsyncTask(params) for params in params_list]
def __enter__(self):
for thread in self.threads:
thread.start()
return self
def __exit__(self, *args):
for thread in self.threads:
if thread.is_alive():
thread.stop()
thread.join() # wait for it to terminate
return None # allows exceptions to be processed normally
params = ['Fee', 'Fie', 'Foe']
with AsyncTaskListContextManager(params) as task_list:
for _ in range(5):
sleep(1)
print('leaving task list context')
print('end-of-script')
</code></pre>
<p>输出:</p>
<pre class="lang-none prettyprint-override"><code>task 'Fee' running
task 'Fie' running
task 'Foe' running
task 'Foe' running
task 'Fee' running
task 'Fie' running
... etc
task 'Fie' running
task 'Fee' running
task 'Foe' running
leaving task list context
task 'Foe' stopped
task 'Fie' stopped
task 'Fee' stopped
end-of-script
</code></pre>