4

我想要做的是能够调用具有多个线程的函数并获得它们的结果。

我有以下代码：

（这是一个例子，实际代码不会简单地将 str 转换为 int）

from threading import Thread
import time
import Queue

#an example - actual code connects to a server
def str_to_int(arg, queue):
    result = 0
    result = int(arg)
    #sleep to check that they happen at once.
    time.sleep(10)
    queue.put(result)

def combine():
    q1 = Queue.Queue()
    q2 = Queue.Queue()
    q3 = Queue.Queue()

    t1 = Thread(target = str_to_int, args=("111", q1))
    t2 = Thread(target = str_to_int, args=("222", q2))
    t3 = Thread(target = str_to_int, args=("333", q3))

    t1.start()
    t2.start()
    t3.start()

    t1.join()
    t2.join()
    t3.join()

    return (q1.get(),q2.get(),q3.get())

print combine()

此代码有效。我得到了预期的结果：

>>> 
(111, 222, 333)

但是，必须有更好的方法来做到这一点。我计划拥有比 3 更多的线程，但即使我只使用 3 - 它看起来也很丑陋。

编辑：我需要能够知道哪个结果来自哪个线程（即：我给函数的参数/参数）

4

2 回答 2

7

这里有一些建议：

  1. Queue是线程安全的，所以使用 1 个队列来传递结果。
  2. 您可以在一个循环中创建所有线程并使用您的队列来传递结果。您不需要为每个线程都有显式变量。

因此，您的代码可能如下所示：

def str_to_int(arg, queue):
    result = int(arg)
    queue.put({arg: result})

def combine():
    arguments = ('111', '222', '333')
    q = Queue.Queue()
    threads = []

    for argument in arguments:
        t = Thread(target=str_to_int, args=(argument, q))
        t.start()
        threads.append(t)

    for t in threads:
        t.join()

    return [q.get() for _ in xrange(len(arguments))]
于 2012-08-16T11:31:39.120 回答
0

我需要能够知道哪个结果来自哪个线程（即：我给函数的参数/参数）

我使用这样的函数来跟踪哪个结果来自哪个任务：

from threading import Thread
import typing

def run_together(tasks: typing.Dict[typing.Hashable, tuple],
                 except_errors: tuple = None) -> dict:
    """
    :param tasks: a dict of task keys matched to a tuple of callable and its arguments
    <pre>tasks = {'task1': (add, 1, 2), 'task2': (add, 3, 4)}</pre>

    :param except_errors: a tuple of errors that should be catched. Defaults to all errors 
    :return: a dictionary of results with the same keys as `tasks` parameter
    <pre>results = {'task1': 3, 'task2': 7}</pre>
    """
    # catch all exceptions by default
    if not except_errors:
        except_errors = (Exception,)
    threads = []
    results = dict()

    def save_results(f, key):
        def wrapped(*args, **kwargs):
            try:
                result = f(*args, **kwargs)
            except except_errors as e:
                result = e
            results[key] = result

        return wrapped

    for key, (func, *args) in tasks.items():
        thread = Thread(
            target=save_results(func, key),
            args=args
        )
        thread.start()
        threads.append(thread)

    for t in threads:
        t.join()

    return results

tasks参数是一个键字典和一个可调用元组及其参数。如果需要，您可以修改save装饰器以返回 a list。

所以你可以这样做：

def add(first, second):
    return first + second

tasks = {f'result:{n}': (add, n, 1) for n in range(4)}
results = run_together(tasks)
print(results)

这使：

{'result:0': 1, 'result:1': 2, 'result:2': 3, 'result:3': 4}
于 2019-03-03T21:26:35.853 回答