如何管理python线程结果？

Question

我正在使用此代码:

def startThreads(arrayofkeywords):
    global i
    i = 0
    while len(arrayofkeywords):
        try:
            if i<maxThreads:
                keyword = arrayofkeywords.pop(0)
                i = i+1
                thread = doStuffWith(keyword)
                thread.start()
        except KeyboardInterrupt:
            sys.exit()
    thread.join()

对于python中的线程,我几乎已经完成了所有工作,但我不知道如何管理每个线程的结果,在每个线程上我都有一个字符串数组作为结果,如何安全地将所有这些数组加入一个？因为,如果我尝试写入全局数组,两个线程可能同时写入.

Answer 1

使用Queue.Queue本质上是线程安全的实例.每个线程在结束时都可以.put将结果发送到该全局实例,并且主线程(当它知道所有工作线程完成时,.join例如在@ unholysampler的答案中),可以循环.get每个结果,并使用每个结果到.extend"整体结果"列表,直到队列清空为止.

编辑:您的代码还有其他大问题 - 如果最大线程数小于关键字数量,它将永远不会终止(您尝试为每个关键字启动一个线程 - 永远不会更少 - 但如果您已经开始了你永远循环的最大数字,没有进一步的目的).

相反,请考虑使用线程池,类似于此配方中的线程池,除了代替排队callables,您将对关键字进行排队 - 因为您想要在线程中运行的可调用在每个线程中是相同的,只是变化争论.当然,可调用的内容将被更改为从传入任务队列(带.get)中剥离某些内容,并.put在完成时将结果列表剥离到传出结果队列.

要终止N个线程,你可以在所有关键字之后使用.putN个"哨兵"(例如None,假设没有关键字可以None):如果它刚刚拉出的"关键字",则线程的可调用将退出None.

通常情况下,Queue.Queue提供了在Python中组织线程(和多处理!)体系结构的最佳方法,它们是我在我指向的配方中的通用,或者更像我在最后两段中建议你的用例.

Answer 2

首先,您实际上需要保存所有这些thread对象以调用join()它们.如上所述,您只保存最后一个,然后仅保存没有异常.

执行多线程编程的一种简单方法是为每个线程提供运行所需的所有数据,然后让它不写入该工作集之外的任何内容.如果所有线程都遵循该指南,则它们的写入不会相互干扰.然后,一旦线程完成,主线程只将结果聚合到一个全局数组中.这被称为"fork/join parallelism".

如果您继承Thread对象,则可以为其提供空间来存储该返回值,而不会干扰其他线程.然后你可以做这样的事情:

class MyThread(threading.Thread):
    def __init__(self, ...):
        self.result = []
        ...

def main():
    # doStuffWith() returns a MyThread instance
    threads = [ doStuffWith(k).start() for k in arrayofkeywords[:maxThreads] ]
    for t in threads:
        t.join()
        ret = t.result
        # process return value here

编辑:

看了一下之后,似乎上面的方法不是在Python中执行线程的首选方法.以上是线程的Java-esque模式.相反,你可以这样做:

def handler(outList)
    ...
    # Modify existing object (important!)
    outList.append(1)
    ...

def doStuffWith(keyword):
    ...
    result = []
    thread = Thread(target=handler, args=(result,))
    return (thread, result)

def main():
    threads = [ doStuffWith(k) for k in arrayofkeywords[:maxThreads] ]
    for t in threads:
        t[0].start()
    for t in threads:
        t[0].join()
        ret = t[1]
        # process return value here