为了并行化一个简单的for循环，joblib为多处理的原始使用带来了很多价值。不仅是简短的语法，还包括在迭代非常快的时候(以消除开销)进行透明的迭代聚集，或者捕获子进程的回溯，以获得更好的错误报告。

免责声明:我是joblib的原作者。

由于@iuryxavier

from multiprocessing import Pool
from multiprocessing import cpu_count


def add_1(x):
    return x + 1

if __name__ == "__main__":
    pool = Pool(cpu_count())
    results = pool.map(add_1, range(10**12))
    pool.close()  # 'TERM'
    pool.join()   # 'KILL'

为了并行化一个简单的for循环，joblib为多处理的原始使用带来了很多价值。不仅是简短的语法，还包括在迭代非常快的时候(以消除开销)进行透明的迭代聚集，或者捕获子进程的回溯，以获得更好的错误报告。

免责声明:我是joblib的原作者。

Dask期货;我很惊讶至今还没有人提起这件事……

from dask.distributed import Client

client = Client(n_workers=8) # In this example I have 8 cores and processes (can also use threads if desired)

def my_function(i):
    output = <code to execute in the for loop here>
    return output

futures = []

for i in <whatever you want to loop across here>:
    future = client.submit(my_function, i)
    futures.append(future)

results = client.gather(futures)
client.close()

并行处理的一个非常简单的例子是

from multiprocessing import Process

output1 = list()
output2 = list()
output3 = list()

def yourfunction():
    for j in range(0, 10):
        # calc individual parameter value
        parameter = j * offset
        # call the calculation
        out1, out2, out3 = calc_stuff(parameter=parameter)

        # put results into correct output list
        output1.append(out1)
        output2.append(out2)
        output3.append(out3)

if __name__ == '__main__':
    p = Process(target=pa.yourfunction, args=('bob',))
    p.start()
    p.join()

使用Ray有很多优点:

除了多个核(使用相同的代码)之外，还可以在多台机器上并行。 通过共享内存(和零拷贝序列化)有效地处理数值数据。 具有分布式调度的高任务吞吐量。 容错。

在本例中，您可以启动Ray并定义一个远程函数

import ray

ray.init()

@ray.remote(num_return_vals=3)
def calc_stuff(parameter=None):
    # Do something.
    return 1, 2, 3

然后并行地调用它

output1, output2, output3 = [], [], []

# Launch the tasks.
for j in range(10):
    id1, id2, id3 = calc_stuff.remote(parameter=j)
    output1.append(id1)
    output2.append(id2)
    output3.append(id3)

# Block until the results have finished and get the results.
output1 = ray.get(output1)
output2 = ray.get(output2)
output3 = ray.get(output3)

要在集群上运行相同的示例，唯一需要更改的行是对ray.init()的调用。相关文档可以在这里找到。

请注意，我正在帮助开发雷。