在Python中使用多處理的對象的共享數組

Question

我想在python程序的不同進程之間創建一個共享對象（類的實例）數組。這些對象中的每一個都將在程序中修改。爲此我使用的多爲：

import multiprocessing 
import numpy as np 
sh = multiprocessing.RawArray (ctypes.py_object, 10) 
f = np.frombuffer(sh, dtype=object) 

，我得到的錯誤是：

Traceback (most recent call last): 
    File "<pyshell#14>", line 1, in <module> 
    f = np.frombuffer(sh, dtype=object) 
ValueError: cannot create an OBJECT array from memory buffer 

現在的問題是，首先，這是一般解決這個問題的正確方法第二，在上面的代碼中我的錯誤是什麼？先謝謝你。

Answer 1

ctypes.py_object代表C中的PyObject *。它是一個指向struct的指針，它表示一個Python對象，駐留在進程的私有內存中，並且包含更多指針。另一個進程無法訪問它;試圖在進程之間共享指針是沒有用的。請參閱this answer by Alex Martelli。

Answer 2

您可能想要使用multiprocessing.Queue，您可以在其中轉儲對象而不用擔心類型。它也是線程安全的和流程安全的。

下面是一個簡單的隊列示例，用於簡化生產者 - 消費者問題（original source，比薩餅對我有點幫助）。

from multiprocessing import Process, Queue 

class Pizza(object): 
    def __init__(self, pizza_num): 
     self.pizza_num = pizza_num 
     self.num_slices = 8 

sentinel = "NO PIZZA" 

def producer(initial_num_pizzas, total_num_pizzas, q): 
    """Cooks Pizzas to be consumed and waits for the consumer to finish eating.""" 
    print("Producer: I am cooking %s Pizzas and putting them on the Queue!"%(total_num_pizzas-initial_num_pizzas)) 
    for i in range(q.qsize(), total_num_pizzas): 
     print("Producer: Behold, for I have cooked Pizza no. %s"%i) 
     q.put(Pizza(i)) 
    q.put(sentinel) 

def consumer(q): 
    """Consumes some Pizza. In this case, all it does is set the number of slices to 0.""" 
    while True: 
     pizza = q.get() 
     pizza.num_slices = 0 
     if pizza == sentinel: 
      break 
     print("Comsumer: Pizza no. %s was found! It has %s slices, yum!"%(pizza.pizza_num, pizza.num_slices)) 

if __name__ == '__main__': 
    q = Queue() 
    total_num_pizzas = 10 
    initial_num_pizzas = 4 
    ## Let's add some Pizzas beforehand: 
    for i in range(0, initial_num_pizzas): 
     q.put(Pizza(i)) 
    print("Main: I have precooked %s Pizzas."%q.qsize()) 

    producer_proc = Process(target=producer, args=(initial_num_pizzas, total_num_pizzas, q)) 
    consumer_proc = Process(target=consumer, args=(q,)) 
    producer_proc.start() 
    consumer_proc.start() 

    q.close() ## Shop is closed, no more Pizzas will be added to Queue! 
    q.join_thread() 

    producer_proc.join() 
    consumer_proc.join() 

下面是一個輸出示例。如果運行它，生產者和消費者打印語句可能會以不同方式進行交錯，因爲並行進程的非確定性執行。

Main: I have precooked 4 Pizzas. 
Producer: I am cooking 6 Pizzas and putting them on the Queue! 
Producer: Behold, for I have cooked Pizza no. 4 
Producer: Behold, for I have cooked Pizza no. 5 
Producer: Behold, for I have cooked Pizza no. 6 
Producer: Behold, for I have cooked Pizza no. 7 
Comsumer: Pizza no. 0 was found! It has 8 slices, yum! 
Comsumer: Pizza no. 1 was found! It has 8 slices, yum! 
Producer: Behold, for I have cooked Pizza no. 8 
Comsumer: Pizza no. 2 was found! It has 8 slices, yum! 
Producer: Behold, for I have cooked Pizza no. 9 
Comsumer: Pizza no. 3 was found! It has 8 slices, yum! 
Comsumer: Pizza no. 4 was found! It has 8 slices, yum! 
Comsumer: Pizza no. 5 was found! It has 8 slices, yum! 
Comsumer: Pizza no. 6 was found! It has 8 slices, yum! 
Comsumer: Pizza no. 7 was found! It has 8 slices, yum! 
Comsumer: Pizza no. 8 was found! It has 8 slices, yum! 
Comsumer: Pizza no. 9 was found! It has 8 slices, yum! 

請注意，您應該use Sentinels to mark the end of your Queue。我在這裏使用了「NO PIZZA」，但他們可以做任何事情。