等待列表中的特定线程

Question

我有 3 个线程：A、B 和 C。

它们在 threads 列表中，并与

同时触发

A.start()
B.start()
C.start()

我不知道哪个线程会先完成。

我需要在每个线程完成时触发它的动作，就像这样：

for t in threads:
    t.join()
    action()

此代码在这种情况下不正确，因为：我必须等待第一个 A 线程在检查 B 之前完成，并等待 B 完成以检查 C 等等。 .

如何以独立的方式等待每个线程，好像“.join()”是一个事件？

Answer 1

您的问题遗漏了许多细节，所以我猜测并提出以下方法来演示一种方法：

from random import random
from threading import Thread
from time import sleep

def callback(name):
    print('thread {} finished'.format(name))

A = Thread(target=lambda: sleep(random()))
B = Thread(target=lambda: sleep(random()))
C = Thread(target=lambda: sleep(random()))

A.start()
B.start()
C.start()

threads = {'A': A, 'B': B, 'C': C}
while threads:
    for name, thread in sorted(threads.items()):
        if not thread.is_alive():
            callback(name)
            threads.pop(name)
            break

print('done')

典型输出：

thread C finished
thread A finished
thread B finished
done

Answer 2

您正在寻找的是一种在线程组上等待的方法，而 Python 的 API 没有提供这种方法。所以，你必须模拟它。

---

最简单但效率最低的方法就是忙轮询线程：

while threads:
    for thread in threads:
        thread.join(timeout=0.1)
        if not thread.is_alive():
            action(thread)
            threads.remove(thread)
            break
    time.sleep(0.1)

---

更好的解决方案是使用一个同步对象，所有线程都可以在它们退出之前发出信号。你可以用一个简单的Event：

while threads:
    evt.wait()
    evt.clear()
    for thread in threads:
        if not thread.is_alive():
            thread.join()
            action(thread)
            threads.remove(thread)
            break

但是，这与任何自重置事件具有相同的潜在竞争条件：您可能会错过触发器。在某些情况下，您可以仔细研究所有细节并确定比赛不会造成任何伤害。如果没有，您必须使用稍微复杂一些的东西，例如 Condition。

---

或者您可以跳到更高的级别并使用Queue，线程可以在退出之前使用q.put(self)。那么主线程就容易搞定了：

while threads:
    thread = q.get()
    thread.join()
    action(thread)
    threads.remove(thread)

---

但值得考虑的是，您是否真的需要 3 个线程函数，或者 3 个可以在池或执行器上运行的任务函数，或者甚至是 1 个具有 3 个不同参数的任务函数。

例如，对于具有 3 组 args 的单个函数，很难比这更简单：

with multiprocessing.dummy.Pool(3) as pool:
    for result in pool.imap_unordered(func, [arg1, arg2, arg3]):
        action(result)

或者，对于 3 个功能：

with concurrent.futures.ThreadPoolExecutor(max_workers=3) as ex:
    fs = [ex.submit(func) for func in funcs]
    for f in concurrent.futures.as_completed(fs):
        action(f.result())

或者您甚至可以将action 添加到期货上：

with concurrent.futures.ThreadPoolExecutor(max_workers=3) as ex:
    for func in funcs:
        ex.submit(func).add_done_callback(action)