Python 다중 처리 : 다음 결과 검색

Question

multiprocessing 패키지를 Python 3.6에서 사용하여 최대 100 개의 작업 세트를 실행하고 그 중 최대 4 개를 동시에 실행할 수 있습니다.

1. 다음 작업을 풀에서 수확하고 모든 작업이 성공 또는 실패 할 때까지 해당 반환 값을 반복적으로 처리합니다.
2. 주어진 작업에서 예외를 발생시키지 않으면 치명적이지 않으므로 다른 작업의 결과에 계속 액세스 할 수 있습니다.

풀에 제출 된 작업의 순서를 유지할 필요가 없습니다 (즉, 대기열이 필요하지 않음). 총 작업 수 (위의 "100")는 엄청나게 큰 것은 아닙니다. 한 번에 모두 제출하고 작업자가 대기 상태가 될 때까지 대기 시키도록하겠습니다.

나는 multiprocessing.Pool이 이것에 잘 맞을 것이라고 생각했지만 반복적으로 호출 할 수있는 "다음 결과 얻기"방법을 찾을 수없는 것 같습니다.

프로세스 관리 기본 요소에서 벗어나야 할 부분입니까? 또는 Pool (또는 제가 누락 된 다른 문제)이 워크 플로를 지원할 수 있습니까?

컨텍스트에서 각 작업자는 원격 프로세스를 호출하는데 몇 분이 걸릴 수 있으며 동시에 N 개의 작업을 처리 할 수있는 용량을 가지고 있습니다 (위의 구체화 된 예제에서 "4").

Answer 1

내가 (대신 4 & (100)의 2 명 근로자 & 6 개 작업을 사용하여 표시) 다음과 같은 패턴을 함께했다 : 나는 거

Start: 0 
Start: 1 
Start: 2 
Done: 1 
Start: 3 
Exception: Two is bad 
Start: 4 
Start: 5 
Done: 3 
Done: 4 
Done: 5 
Done: 0 

을 :

import random 
import time 
from multiprocessing import Pool, TimeoutError 
from queue import Queue 


def worker(x): 
    print("Start: {}".format(x)) 
    time.sleep(5 * random.random()) # Sleep a random amount of time 
    if x == 2: 
     raise Exception("Two is bad") 
    return x 


if __name__ == '__main__': 

    with Pool(processes=2) as pool: 
     jobs = Queue() 
     for i in range(6): 
      jobs.put(pool.apply_async(worker, [i])) 

     while not jobs.empty(): 
      j = jobs.get(timeout=1) 
      try: 
       r = j.get(timeout=0.1) 
       print("Done: {}".format(r)) 
      except TimeoutError as e: 
       jobs.put(j) # Not ready, try again later 
      except Exception as e: 
       print("Exception: {}".format(e)) 

꽤 잘 작동하는 것 같다 내가 나를 위해 큐를 관리하는 일반적인 유틸리티를 만들 수 있는지 확인하십시오.

내가 생각하는 주요 단점은 완성 된 작업이 잠시 동안 눈에 띄지 않을 수 있지만 완료되지 않은 작업은 폴링되고 시간이 초과 될 수 있다는 것입니다. 그것을 회피하려면 아마 콜백을 사용해야 할 것입니다. 문제가 커지면 아마 내 앱에 추가 할 것입니다.