在Python中使用多处理时不执行子进程

Question

我现在正在测试下面的代码以并行计算数组，但似乎无法执行长时间函数。我的Python版本：3.7.4，操作系统：win 10。

from multiprocessing import Pool, Lock, Array
import os, time    
def long_time_task(i,array,lock):
print('Run task %s (%s)...' % (i, os.getpid()))
start = time.time()
total_count = 0
for k in range(5*10**7): total_count += 1
total_count += i
lock.acquire()
array[i] = total_count
lock.release()
end = time.time()
print('Task %s runs %0.2f seconds.' % (i, (end - start)))

def mainFunc():
print('Parent process %s.' % os.getpid())
p = Pool()
array = Array('f', 20)
lock = Lock()
for i in range(20): p.apply_async(long_time_task, args=(i,array,lock))
print('Waiting for all subprocesses done...')
p.close()
p.join()
print('All subprocesses done.')

if __name__ == '__main__':    
mainFunc()

Answer 1

您的代码有几个问题：

1. apply_async方法返回一个需要等待的结果对象。

2. 不能将正常的多处理Array或Lock对象传递给Pool方法，因为它们不能被pickle。您可以改用manager对象。

试试这个：

from multiprocessing import Pool, Lock, Array, Manager
import os, time
def long_time_task(i,array,lock):
print('Run task %s (%s)...' % (i, os.getpid()))
start = time.time()
total_count = 0
for k in range(5*10**7): total_count += 1
total_count += i
lock.acquire()
array[i] = total_count
lock.release()
end = time.time()
print('Task %s runs %0.2f seconds.' % (i, (end - start)))

def mainFunc():
print('Parent process %s.' % os.getpid())
p = Pool()
m = Manager()
array = m.Array('f', [0] * 20)
lock = m.Lock()
results  = [p.apply_async(long_time_task, args=(i,array,lock)) for i in range(20)]
[result.get() for result in results]
print('Waiting for all subprocesses done...')
p.close()
p.join()
print('All subprocesses done.')

if __name__ == '__main__':
mainFunc()

你也可以这样简化它，但我不确定这是否适用于你的情况：

import array
import os
import time
from multiprocessing import Pool

def long_time_task(i):
print(f'Run task {i} ({os.getpid()})...')
start = time.time()
total_count = 0
for k in range(5 * 10 ** 7):
total_count += 1
total_count += i
end = time.time()
print(f'Task {i} runs {end - start:.2f} seconds.')
return total_count

def main():
print('Parent process %s.' % os.getpid())
a = array.array('d', range(20))
r = range(20)
with Pool() as pool:
for idx, result in zip(r, pool.map(long_time_task, r)):
a[idx] = result
print(a)
print(f'All subprocesses done.')

if __name__ == '__main__':
main()