在python中并行执行任务
我使用的是python 2.7，我有一些代码看起来像这样：
task1()
task2()
task3()
dependent1()
task4()
task5()
task6()
dependent2()
dependent3()这里唯一的依赖关系如下: dependent1需要等待任务1-3，dependent2需要等待任务4-6，dependent3需要等待依赖1-2...可以这样做:首先并行运行全部6个任务，然后并行运行前两个从属任务。然后是最终的依赖项
我更喜欢让尽可能多的任务并行运行，我已经在谷歌上搜索了一些模块，但我希望避免使用外部库，并且不确定队列线程技术如何解决我的问题(也许有人可以推荐一个好的资源？)
回答 3
Stack Overflow用户
发布于 2011-11-23 20:40:48
内置的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等待线程结束。
Stack Overflow用户
发布于 2019-02-07 05:21:25
如果你愿意尝试一下外部库，你可以用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您还可以在任务之间传递实际的python对象，方法是使用任务内部的参数并返回结果(例如，说"return value“而不是上面的" pass”)。
使用"pip install ray“，上面的代码可以在一台机器上开箱即用，也很容易在集群上并行应用程序，无论是在云中还是在您自己的自定义集群中，请参阅https://ray.readthedocs.io/en/latest/autoscaling.html和https://ray.readthedocs.io/en/latest/using-ray-on-a-cluster.html)。如果以后您的工作负载增加，这可能会派上用场。
免责声明:我是Ray的开发者之一。
Stack Overflow用户
发布于 2011-11-23 19:21:31
看看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)霍普，这就是你一直在找的。
https://stackoverflow.com/questions/8241099
复制相似问题
腾讯云开发者
