Python多进程：进程无法工作

Question

我想把我的计算工作分配给多个CPU，所以我选择了多进程处理。然而，结果并不是我想要的。

import numpy as np
from multiprocessing import Process

def func(begin,end):
    print('*'*5)
    print('begin=%d' %(begin))
    for i in range(begin,end):
        for j in range(10):
            myarray[i][j]=1

myarray=np.zeros((12,10))
print(myarray)
for i in range(4):
    begin=i*3
    end=(i+1)*3
    p=Process(target=func,args=(begin,end,))
    p.start()

print('*'*5)
print(myarray)

我认为myarray应该全是1。但它却一直没有改变。为什么？func函数不会改变myarray的元素吗？ 我尝试了来自这个链接的示例。

from multiprocessing import Process
def f(name):
    print('hello',name)

p=Process(target=f,args=('bob',))
p.start()

屏幕上什么也没有显示。为什么？ 我该如何使用Python完成计算？有人能提供一个充分利用多CPU的方法吗？

- dudu

---

第二个代码片段输出：(hello，bob) - Nir Alfasi

@alfasin 看起来应该可以。但我在win7上使用Python 3.4.3测试了该代码，并在CentOS6.5的虚拟机中使用Python 2.6.6测试了相应的代码（将print函数转换为语句），但两者都没有显示任何内容。 - dudu

它适用于我在Mac使用Python3.4。 - Nir Alfasi

@alfasin 抱歉，第二个在win7上使用python3.4.3无法运行。这是我用来测试简单程序的平台。但是在centos上使用python 2.6.6可以正常运行。我再次检查了一下。虽然我不知道原因，但现在就是这种情况。 - dudu

Answer 1

那里有两个问题：

1. 当你在最后打印数组时，你如何知道进程已经完成？你需要在每个进程上调用join()以确保它们已经完成。

2. 每个进程都有一个副本 "myarray"。如果你想要通信多个进程，你需要使用一个Queue或一个Pipe。请查看讲述进程间交换数据的文档

这是一个使用你发布的基础的工作示例（它不打算快速，只是展示如何进行通信）：

from multiprocessing import Process, freeze_support, Queue

def func(my_id, q, begin, end):
    global myarray
    print('Process %d has range: %d - %d' % (my_id, begin, end))
    for i in range(begin,end):
        q.put((i, i * 2))

if __name__ == "__main__":
    freeze_support()

    q = Queue()
    processes = []
    myarray=[0] * 12
    print("At the beginning the array is ", myarray)

    for i in range(4):
        begin = i*3
        end = (i+1)*3
        p = Process(target=func, args=(i, q, begin, end))
        p.start()
        processes.append(p)

    for p in processes:
        p.join()

    while not q.empty():
        (index, value) = q.get()
        myarray[index] = value

    print("At the end the array is ", myarray)

尝试将行 p.join() 更改为 pass，看看会发生什么 :)

- Daniel

---

是的，join 是必须的。如果我用 pass 替换 p.join()，一些值会改变而另一些则不会。 - dudu

Answer 2

这是因为 print(myarray) 语句在进程结束之前执行了。你应该等待进程结束，然后再执行此语句。请注意以下的 thread.join() 语句。

import numpy as np
import threading

lock = threading.RLock()
thread_list = []
def func(begin,end):
    print('*'*5)
    print('begin=%d' %(begin))
    for i in range(begin,end):
        for j in range(10):
            with lock:
                myarray[i][j]=1

myarray=np.zeros((12,10))
print(myarray)
for i in range(4):
    begin=i*3
    end=(i+1)*3
    p=threading.Thread(target=func,args=(begin,end,))
    p.start()
    thread_list.append(p)

print('*'*5)
for thread in thread_list:
    thread.join()
print(myarray)

- peihuadeng