加入Python中的一个线程

Question

我有一个Python程序，其中包括一个主线程和两个或更多其他线程（也许数量不重要）。我希望让主线程休眠直到其中一个其他线程完成。使用轮询很容易实现（通过调用t.join(1)并等待每个线程t一秒钟）。

是否可以在不进行轮询的情况下完成？

SOMETHING_LIKE_JOIN(1, [t1, t2])

其中t1和t2是线程.Thread对象。该调用必须执行以下操作：休眠1秒钟，但一旦t1或t2中的一个完成，就立即唤醒。与POSIX select(2)调用具有两个文件描述符非常相似。

- ilya

---

也许是条件对象？主线程等待条件对象。t1和t2在返回之前调用condition_obj.notify()。先返回的线程会唤醒主线程。以下是条件对象的文档链接：link - Ma Tingchen

我在这里看到了竞态条件：想象一下 t1 完成并调用“notify”，主线程醒来，做一些事情后又要再次进入睡眠状态（等待 t2、t3、t4 等）。可能会发生的情况是，在主线程“做一些事情”的过程中，一些甚至所有的 t2、……、t4 线程都会完成任务并调用“notify”。然后，主线程将调用“wait”并永久沉睡，因为没有人再唤醒它了。 - ilya

没有竞争条件。条件对象包含一个RLock。稍后我会发布一个例子。 - Ma Tingchen

Answer 1

一种解决方案是使用; 提供了一个几乎与相同的API，但由线程支持，因此可以免费获得线程池。

例如，您可以执行以下操作：

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

会在任务完成时立即返回结果，而不管哪个任务先完成。

如果您使用的是Python 3.2或更高版本（或安装了旧版本Python的 PyPI模块），则可以通过从创建一个或多个，然后使用和，或使用来将异类任务函数泛化为相似效果。

- ShadowRanger

Answer 2

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

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来确保该条件永远不会通知主线程，直到主线程开始等待。我不喜欢它的工作方式，但我还没有找到更好的解决方案...

- Ma Tingchen

---

请在锁、条件等语句中使用with语句。这样可以避免忘记释放锁的风险，并防止异常绕过release。此外，这种方式更加简洁，run函数的最后四行代码可以改为：with condition:、condition.notify()、print('thread_%d' % idx)。 - ShadowRanger

谢谢您的建议。我已经编辑了run()函数以及main()函数中的while循环块。 - Ma Tingchen

Answer 3

你可以创建一个Thread类，主线程保持对它的引用。这样，你就可以检查线程是否已经完成并使你的主线程再次轻松地继续执行。

如果这不帮助您，我建议您看一下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)

- Rotsechs96

---

不行，这个问题是关于“无轮询”解决方案的，即不能调用“sleep(1)”。 - ilya