更新可能なキューを使ったマルチプロセッシング

Question

私は、マルチプロセッシング・キューを使用するPythonでプログラムを書く方法を理解しようとしています。

私は、複数のサーバーを持っており、そのうちの一つは、このを使用してリモートキューを提供します：

from multiprocessing.managers import BaseManager 
import Queue 
import daemonme 

queue = Queue.Queue() 

class QueueManager(BaseManager): 
    pass 

daemonme.createDaemon() 
QueueManager.register('get_job', callable=lambda:queue) 
m = QueueManager(address=('', 50000), authkey='') 
s = m.get_server() 
s.serve_forever() 

今、私は、このリモート・キューのオフジョブを処理するために私のデュアルXeonプロセッサ、クアッドコアサーバを使用したいです。仕事はお互いに完全に独立しています。だから私が8コアを持っているなら、キューからジョブを選んで処理し、次のプロセスに戻る7つのプロセスを開始したいと思います。 7つのプロセスのそれぞれがこれを行いますが、私はこのプログラムの構造を頭で覆うことはできません。

誰も私にこの基本的な構造に関する知識を提供することはできますか？

ありがとうございます。

Answer 1

master-slave（別名farmer-worker）パターンを使用する必要があります。最初のプロセスはマスターになり、ジョブが作成されます。それ

1. はキュー
2. がキューに

をジョブを書い

開始スレーブ・プロセスが継続的にキューから読み出され、実行パラメータとしてキューを渡し、7つのスレーブ・プロセスを作成し、作成しますジョブ（キューから停止メッセージを受け取るまで）。このシナリオでは、AFAICTを使用する必要はありません。

Answer 2

労働者のpool（段落17.6.2.9）よりもmanager（段落17.6.2.7） からキューを盗んする方法ドキュメントへのルックは、それぞれにキューを渡し7つのジョブを起動します。あなたの実装は*リモート*キューを管理する方法を

from multiprocessing.managers import BaseManager 
import random 

class Producer(): 
def __init__(self): 
    BaseManager.register('queue') 
    self.m = BaseManager(address=('hostname', 50000), authkey='jgsjgfdjs') 
    self.m.connect() 
    self.cm_queue = self.m.queue() 
    while 1: 
     time.sleep(random.randint(1,3)) 
     self.cm_queue.put(<PUT-HERE-JOBS>) 

from multiprocessing.managers import BaseManager 
import time 
import random 
class Consumer(): 
def __init__(self): 
    BaseManager.register('queue') 

    self.m = BaseManager(address=('host', 50000), authkey='jgsjgfdjs') 
    self.m.connect() 
    self.queue = self.m.queue() 
    while 1: 
     <EXECUTE(job = self.queue.get())> 


from multiprocessing.managers import BaseManager, Queue 
class Manager(): 

def __init__(self): 

    self.queue = QueueQueu() 

    BaseManager.register('st_queue', callable=lambda:self.queue) 

    self.m = BaseManager(address=('host', 50000), authkey='jgsjgfdjs') 
    self.s = self.m.get_server() 

    self.s.serve_forever()