재생 대기열을 사용하는 다중 처리

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() 

는 지금이 원격 큐의 해제 작업을 처리하기 위해 내 듀얼 제온 쿼드 코어 서버를 사용하고 싶습니다. 작업은 서로 완전히 독립적입니다. 따라서 8 개의 코어가 있다면 큐에서 작업을 선택하고 처리 한 다음 다음 작업으로 돌아가는 7 개의 프로세스를 시작하고 싶습니다. 7 가지 프로세스가 모두이 작업을 수행하지만,이 프로그램의 구조를 감싸는 데 머리를 쓰지는 못합니다.

누구나 내게이 기본 구조에 대한 교육을받은 아이디어를 제공 할 수 있습니까?

미리 감사드립니다.

Answer 1

master-slave (일명 farmer-worker) 패턴을 사용해야합니다. 초기 프로세스는 마스터가되며 작업을 만듭니다. 그것은

1. 큐에 파라미터로서 연속적 큐로부터 판독

슬레이브 프로세스 작업을 작성

시작 큐 전달

7 개 슬레이브 프로세스를 생성 큐를 생성하고, 수행 작업 (대기열에서 중지 메시지를받을 때까지). 이 시나리오에서는 Manager 객체를 사용할 필요가 없습니다.

Answer 2

노동자의 pool (단락 17.6.2.9)와보다 manager (단락 17.6.2.7) 에서 큐를 retreive하는 방법을 문서로 봐 각 하나에 큐를 통과 (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()