加入 Python 中的多个线程之一

Question

我有一个带有一个主线程的 python 程序，假设有 2 个其他线程（或者甚至更多，可能没关系）。我想让主线程休眠，直到其他线程之一完成。轮询很容易做到（通过调用 t.join(1) 并为每个线程 t 等待一秒钟）。

是否可以不进行轮询，仅通过

SOMETHING_LIKE_JOIN(1, [t1, t2])

其中 t1 和 t2 是 threading.Thread 对象？该调用必须执行以下操作：休眠 1 秒，但在 t1、t2 之一完成后立即唤醒。与使用两个文件描述符的 POSIX select(2) 调用非常相似。

Answer 1

这里是一个使用条件对象的例子。

from threading import Thread, Condition, Lock
from time import sleep
from random import random


_lock = Lock()


def run(idx, condition):
    sleep(random() * 3)
    print('thread_%d is waiting for notifying main thread.' % idx)
    _lock.acquire()
    with condition:
        print('thread_%d notifies main thread.' % idx)
        condition.notify()


def is_working(thread_list):
    for t in thread_list:
        if t.is_alive():
            return True
    return False


def main():
    condition = Condition(Lock())
    thread_list = [Thread(target=run, kwargs={'idx': i, 'condition': condition}) for i in range(10)]

    with condition:
        with _lock:
            for t in thread_list:
                t.start()

            while is_working(thread_list):
                _lock.release()
                if condition.wait(timeout=1):
                    print('do something')
                    sleep(1)  # <-- Main thread is doing something.
                else:
                    print('timeout')

    for t in thread_list:
        t.join()


if __name__ == '__main__':
    main()

我认为没有您在评论中描述的竞争条件。条件对象包含一个锁。当主线程工作时（例如sleep(1)），它持有锁，没有线程可以通知它，直到它完成它的工作并释放锁。

---

我刚刚意识到前面的示例中存在竞争条件。我添加了一个全局 _lock 以确保在主线程开始等待之前条件永远不会通知主线程。我不喜欢它的工作方式，但我还没有找到更好的解决方案......

Answer 2

一种解决方案是使用multiprocessing.dummy.Pool； multiprocessing.dummy 提供的 API 几乎与 multiprocessing 相同，但由线程支持，因此它可以免费为您提供线程池。

例如，你可以这样做：

from multiprocessing.dummy import Pool as ThreadPool

pool = ThreadPool(2)  # Two workers
for res in pool.imap_unordered(some_func, list_of_func_args):
    # res is whatever some_func returned

multiprocessing.Pool.imap_unordered 在结果可用时返回结果，无论哪个任务先完成。

如果您可以使用 Python 3.2 或更高版本（或为旧版 Python 安装 concurrent.futures PyPI 模块），您可以通过从 ThreadPoolExecutor 创建一个或多个 Futures，然后使用 @987654322 来概括不同的任务函数@ 与return_when=FIRST_COMPLETED，或使用concurrent.futures.as_completed 以获得类似的效果。

Answer 3

你可以创建一个线程类并且主线程保持对它的引用。所以你可以检查线程是否已经结束，让你的主线程很容易再次继续。

如果这对您没有帮助，我建议您查看 Queue 库！

import threading
import time, random


#THREAD CLASS#
class Thread(threading.Thread):

    def __init__(self):
        threading.Thread.__init__(self)

        self.daemon = True
        self.state = False

        #START THREAD (THE RUN METHODE)#
        self.start()

    #THAT IS WHAT THE THREAD ACTUALLY DOES#
    def run(self):

        #THREAD SLEEPS FOR A RANDOM TIME RANGE# 
        time.sleep(random.randrange(5, 10))

        #AFTERWARDS IS HAS FINISHED (STORE IN VARIABLE)#
        self.state = True


    #RETURNS THE STATE#
    def getState(self):

        return self.state


#10 SEPERATE THREADS#
threads = []

for i in range(10):
    threads.append(Thread())

#MAIN THREAD#
while True:

    #RUN THROUGH ALL THREADS AND CHECK FOR ITS STATE#
    for i in range(len(threads)):
        if threads[i].getState():
            print "WAITING IS OVER: THREAD ", i 

    #SLEEPS ONE SECOND#
    time.sleep(1)