从线程获取第一个结果

Question

我有一个由 for 循环组成的函数，处理每个元素需要很长时间。

在这个循环中，我只关心是否找到一个结果，循环执行的顺序无关紧要。

如果没有找到结果，它只需要处理所有元素。

我的代码如下所示：

import random
from time import sleep

class Elements:
    def __init__(self,id):
        self.id = id

    def analyze(self,result):
        sleep(random.random()/2)
        if result > 0.9:
            return True

    def get_result(self):
        sleep(random.random()/2)
        return random.random()

def long_process(element):
    result = element.get_result()
    print element.id,result
    if element.analyze(result) == True:
        return element, result
    results.append(result)

if __name__ == '__main__':
    elements = []
    for n in xrange(10):
        elements.append(Elements(n))

    results = []
    desired = None
    for element in elements:
        desired = long_process(element)
        if desired:
            break
    if desired:
        print desired
    else:
        print sorted(results,reverse=True)[0]

如果在元素中找到所需的元素，它会输出：

0 0.150771147344
1 0.510095601312
2 0.28908099464
3 0.410008760093
4 0.435872688389
5 0.960307534462
(<__main__.Elements instance at 0x02311AD0>, 0.9603075344618438)

如果不是，它会输出这个：

0 0.361404239639
1 0.207105823026
2 0.851923103243
3 0.720462633138
4 0.58273657804
5 0.380394809851
6 0.590615590942
7 0.130172801534
8 0.57736519194
9 0.597338819896
0.851923103243 # the best element 

我正在尝试使用多处理池在单独的线程中处理每个元素。但我没有得到想要的效果。

这是我尝试过的。

def long_process(element):
    result = element.get_result()
    print element.id,result, element.analyze(result)
    if element.analyze(result) == True:
        return element, result
    results.append(result)

if __name__ == '__main__':
    # freeze_support()
    pool = Pool(processes=2)
    results = []
    pool.imap_unordered(long_process,elements)
    pool.close()
    pool.join()
    print sorted(results)[0]

输出：

0 0.257371095047 None
1 0.8533933601 None
2 0.74433029726 None
3 0.138217827732 None
5 1 True
6 0.961344251697 True
4 0.0664381718173 None
7 0.000891701766875 None
9 1 True
8 0.771957314202 None
Traceback (most recent call last):
  File "C:/User/user/Documents/Python/threading_test.py", line 40, in <module>
    print sorted(results)[0]
IndexError: list index out of range

但这有两个问题：

• 它应该在第 5 个元素处停止，并返回它。
• 如果未找到 True，则应将所有结果存储在结果列表中。

它确实以无序的顺序工作（id 2,3,5,6,4...），它得到了结果，但是当分析为真（5 1 True）时它并没有停止， 也没有将结果附加到结果列表中。[IndexError]

我尝试使用 Pool.apply_async，但我也无法获取结果。

Answer 1

使用您的代码有点辛苦，所以我使用apply_async 做了一个示例，函数非常简单：

from multiprocessing.pool import Pool
# habit, you can use a flag instead ...
from threading import Event
import sys

# tell everyone we're done
done = Event()
# aggregate results here
results = []
success = []

不管怎样，每个进程都会返回一个结果：

def is7(x):
    if x == 7:
        return (x, True, )
    return x

这是一个应用于每个结果的回调，当来自apply_async 的结果返回时使用：

def handle_result(x):
    if type(x) is tuple:
        # a tuple is necessarily the result we want, so we're done here.
        done.set()
        # stop pool
        p.terimnate()
        success.append(x)
        return
    results.append(x)

这是我们遍历项目（其中 1000 个）并在每个项目上应用函数 is7 的地方。如果他们中的任何一个找到解决方案，我们将立即停止并通知

p = Pool()
def run_processes():
    for element in range(0, 1000):
        print "trying %s" % str(element)
        p.apply_async(func=is7, args=(element,), callback=handle_result)

    # wait for completion
    p.close()
    p.join()
# start dispatchig tasks to pool
run_processes()
if done.isSet():
    # result may contain more than one, if two success tasks were executed closely
    print "Found result: %s" % str(success)
else:
    print "No results found, here is the list: %s " % str(results)

注意：不能保证循环会在 7 点准确停止，它可能会在 7 点之后停止一些项目，因为任务是异步分派的。