python subclassing multiprocessing.Process

Question

我是新來的python面向對象，我重寫我的現有應用程序作爲面向對象的版本，因爲現在開發人員越來越多，我的代碼變得不可維護。

通常我使用多隊列，但是我從這個例子http://www.doughellmann.com/PyMOTW/multiprocessing/basics.html，我也可以繼承multiprocessing.Process所以我認爲這是一個好主意，發現我寫了一個類來測試像這樣：

代碼：

from multiprocessing import Process 
class Processor(Process): 
    def return_name(self): 
     return "Process %s" % self.name 
    def run(self): 
     return self.return_name() 

processes = [] 


if __name__ == "__main__": 

     for i in range(0,5): 
       p=Processor() 
       processes.append(p) 
       p.start() 
     for p in processes: 
       p.join() 

但是我無法取回這些值，我怎麼能以這種方式使用隊列呢？

編輯：我想獲得返回值，並想在哪裏把Queues()。

Answer 1

子類multiprocessing.Process：

但是我不能取回值，我該如何使用隊列這種方式？

過程需要Queue()接收結果...如何繼承multiprocessing.Process如下的例子...

from multiprocessing import Process, Queue 
class Processor(Process): 

    def __init__(self, queue, idx, **kwargs): 
     super(Processor, self).__init__() 
     self.queue = queue 
     self.idx = idx 
     self.kwargs = kwargs 

    def run(self): 
     """Build some CPU-intensive tasks to run via multiprocessing here.""" 
     hash(self.kwargs) # Shameless usage of CPU for no gain... 

     ## Return some information back through multiprocessing.Queue 
     ## NOTE: self.name is an attribute of multiprocessing.Process 
     self.queue.put("Process idx={0} is called '{1}'".format(self.idx, self.name)) 

if __name__ == "__main__": 
    NUMBER_OF_PROCESSES = 5 

    ## Create a list to hold running Processor object instances... 
    processes = list() 

    q = Queue() # Build a single queue to send to all process objects... 
    for i in range(0, NUMBER_OF_PROCESSES): 
     p=Processor(queue=q, idx=i) 
     p.start() 
     processes.append(p) 

    # Incorporating ideas from this answer, below... 
    # https://stackoverflow.com/a/42137966/667301 
    [proc.join() for proc in processes] 
    while not q.empty(): 
     print "RESULT: {0}".format(q.get()) # get results from the queue... 

在我的機器，這導致...

$ python test.py 
RESULT: Process idx=0 is called 'Processor-1' 
RESULT: Process idx=4 is called 'Processor-5' 
RESULT: Process idx=3 is called 'Processor-4' 
RESULT: Process idx=1 is called 'Processor-2' 
RESULT: Process idx=2 is called 'Processor-3' 
$ 

使用multiprocessing.Pool：

FWIW，我發現子類化multiprocessing.Process的一個缺點是，您無法充分利用multiprocessing.Pool的所有內置優點; Pool給你一個非常好的API，如果你不需要需要你的生產者和消費者代碼通過隊列彼此交談。

你可以做很多隻是一些有創意的返回值...在下面的例子中，我使用了一個dict()從pool_job()封裝輸入和輸出值...

from multiprocessing import Pool 

def pool_job(input_val=0): 
    # FYI, multiprocessing.Pool can't guarantee that it keeps inputs ordered correctly 
    # dict format is {input: output}... 
    return {'pool_job(input_val={0})'.format(input_val): int(input_val)*12} 

pool = Pool(5) # Use 5 multiprocessing processes to handle jobs... 
results = pool.map(pool_job, xrange(0, 12)) # map xrange(0, 12) into pool_job() 
print results 

這導致：

[ 
    {'pool_job(input_val=0)': 0}, 
    {'pool_job(input_val=1)': 12}, 
    {'pool_job(input_val=2)': 24}, 
    {'pool_job(input_val=3)': 36}, 
    {'pool_job(input_val=4)': 48}, 
    {'pool_job(input_val=5)': 60}, 
    {'pool_job(input_val=6)': 72}, 
    {'pool_job(input_val=7)': 84}, 
    {'pool_job(input_val=8)': 96}, 
    {'pool_job(input_val=9)': 108}, 
    {'pool_job(input_val=10)': 120}, 
    {'pool_job(input_val=11)': 132} 
] 

顯然在pool_job()中還有很多其他的改進，比如錯誤處理，但是這說明了要點。 FYI，this answer提供了另一個如何使用multiprocessing.Pool的例子。

Answer 2

返回值Process.run不會去任何地方。您需要將它們發送回父流程，例如使用multiprocessing.Queue（docs here）。

Answer 3

非常感謝大家。

現在繼承人我得到了它做:)

在這個例子中，我使用，因爲我不希望每個ohter之間，但只有與父進程通信的多個queus。

from multiprocessing import Process,Queue 
class Processor(Process): 
    def __init__(self,queue): 
     Process.__init__(self) 
     self.que=queue 
    def get_name(self): 
     return "Process %s" % self.name 
    def run(self): 
     self.que.put(self.get_name()) 



if __name__ == "__main__": 

     processes = [] 
     for i in range(0,5): 
       p=Processor(Queue()) 
       processes.append(p) 
       p.start() 
     for p in processes: 
       p.join() 
       print p.que.get() 

Answer 4

Mike's answer是最好的，但只是爲了完整性我想提一提，我更喜歡收穫出隊列join上下文所以最後一位應該是這樣的：

[proc.join() for proc in processes] # 1. join 

while not q.empty(): # 2. get the results 
    print "RESULT: %s" % q.get()