在python中并行执行任务

Question

我正在使用Python2.7，我有一些代码如下所示：

task1()
task2()
task3()
dependent1()

task4()
task5()
task6()
dependent2()

dependent3()

这里唯一的依赖项如下：dependent1需要等待tasks1-3，dependent2需要等待tasks 4-6，dependent3需要等待dependents1-2。。。可以这样做：首先并行运行全部6个任务，然后并行运行前两个从属任务。。那么最后的依赖

我更喜欢尽可能多的任务并行运行，我搜索了一些模块，但我希望避免使用外部库，不确定队列线程技术如何解决我的问题（也许有人可以推荐一个好的资源？）

Answer 1

看看Gevent。

示例用法：

import gevent
from gevent import socket

def destination(jobs):
    gevent.joinall(jobs, timeout=2)
    print [job.value for job in jobs]

def task1():
    return gevent.spawn(socket.gethostbyname, 'www.google.com')

def task2():
    return gevent.spawn(socket.gethostbyname, 'www.example.com')

def task3():
    return gevent.spawn(socket.gethostbyname, 'www.python.org')

jobs = []
jobs.append(task1())
jobs.append(task2())
jobs.append(task3())
destination(jobs)

希望，这就是你一直在寻找的。

Answer 2

内置的threading.Thread类提供了您所需要的一切：start来启动新线程，以及join来等待线程的结束。

import threading

def task1():
    pass
def task2():
    pass
def task3():
    pass
def task4():
    pass
def task5():
    pass
def task6():
    pass

def dep1():
    t1 = threading.Thread(target=task1)
    t2 = threading.Thread(target=task2)
    t3 = threading.Thread(target=task3)

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

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

def  dep2():
    t4 = threading.Thread(target=task4)
    t5 = threading.Thread(target=task5)

    t4.start()
    t5.start()

    t4.join()
    t5.join()

def dep3():
    d1 = threading.Thread(target=dep1)
    d2 = threading.Thread(target=dep2)

    d1.start()
    d2.start()

    d1.join()
    d2.join()

d3 = threading.Thread(target=dep3)
d3.start()
d3.join()

或者可以使用Queue.join来等待线程结束。

Answer 3

如果您愿意尝试使用外部库，则可以使用Ray优雅地表示任务及其依赖项。这在一台机器上工作得很好，这里的优点是并行性和依赖性可以用Ray比python多处理更容易表达，而且它没有GIL（全局解释器锁）问题，而GIL（全局解释器锁）通常会阻止多线程有效工作。此外，如果将来需要的话，可以很容易地在集群上扩展工作负载。

解决方案如下：

import ray

ray.init()

@ray.remote
def task1():
    pass

@ray.remote
def task2():
    pass

@ray.remote
def task3():
    pass

@ray.remote
def dependent1(x1, x2, x3):
    pass

@ray.remote
def task4():
    pass

@ray.remote
def task5():
    pass

@ray.remote
def task6():
    pass

@ray.remote
def dependent2(x1, x2, x3):
    pass

@ray.remote
def dependent3(x, y):
    pass

id1 = task1.remote()
id2 = task2.remote()
id3 = task3.remote()

dependent_id1 = dependent1.remote(id1, id2, id3)

id4 = task4.remote()
id5 = task5.remote()
id6 = task6.remote()

dependent_id2 = dependent2.remote(id4, id5, id6)

dependent_id3 = dependent3.remote(dependent_id1, dependent_id2)

ray.get(dependent_id3) # This is optional, you can get the results if the tasks return an object

您还可以通过使用任务内部的参数并返回结果（例如说“return value”而不是上面的“pass”）在任务之间传递实际的python对象。

使用“pip install ray”，上述代码可以在一台机器上开箱即用，而且在集群上并行应用程序也很容易，无论是在云中还是在您自己的自定义集群中，请参见https://ray.readthedocs.io/en/latest/autoscaling.html和https://ray.readthedocs.io/en/latest/using-ray-on-a-cluster.html）。如果以后你的工作量增加的话，这可能会派上用场。

免责声明：我是雷的开发者之一。