Python 다 중 프로 세 스 multiprocessing 용법 사례 분석

이 글 의 사례 는 Python 다 중 프로 세 스 multiprocessing 용법 을 서술 하 였 다.여러분 께 참고 하도록 공유 하 겠 습 니 다.구체 적 으로 는 다음 과 같 습 니 다.
mutilprocess 안내
스 레 드 처럼 프로 세 스 를 관리 합 니 다.이것 은 mutilprocess 의 핵심 입 니 다.그 는 threading 과 매우 비슷 하고 다 핵 CPU 에 대한 이 용 률 이 threading 보다 훨씬 좋 습 니 다.
간단 한 생 성 프로 세 스:

import multiprocessing
def worker(num):
  """thread worker function"""
  print 'Worker:', num
  return
if __name__ == '__main__':
  jobs = []
  for i in range(5):
    p = multiprocessing.Process(target=worker, args=(i,))
    jobs.append(p)
    p.start()

현재 프로 세 스 를 확인 하 는 것 은 프로 세 스에 이름 을 붙 이 고 표지 구분 을 편리 하 게 하 며 추적 하 는 것 입 니 다.

import multiprocessing
import time
def worker():
  name = multiprocessing.current_process().name
  print name, 'Starting'
  time.sleep(2)
  print name, 'Exiting'
def my_service():
  name = multiprocessing.current_process().name
  print name, 'Starting'
  time.sleep(3)
  print name, 'Exiting'
if __name__ == '__main__':
  service = multiprocessing.Process(name='my_service',
                   target=my_service)
  worker_1 = multiprocessing.Process(name='worker 1',
                    target=worker)
  worker_2 = multiprocessing.Process(target=worker) # default name
  worker_1.start()
  worker_2.start()
  service.start()

데 몬 은 주 프로그램 이 종료 되 는 것 을 막 지 않 고 자신의 mutilprocess.setDaemon(True)을 하 는 것 입 니 다.데 몬 이 종료 되 기 를 기다 리 고 있 습 니 다.join 을 추가 하려 면 join 은 부동 소수점 수 치 를 입력 할 수 있 습 니 다.n 을 오래 기다 리 면 기다 리 지 않 습 니 다.
데 몬:

import multiprocessing
import time
import sys
def daemon():
  name = multiprocessing.current_process().name
  print 'Starting:', name
  time.sleep(2)
  print 'Exiting :', name
def non_daemon():
  name = multiprocessing.current_process().name
  print 'Starting:', name
  print 'Exiting :', name
if __name__ == '__main__':
  d = multiprocessing.Process(name='daemon',
                target=daemon)
  d.daemon = True
  n = multiprocessing.Process(name='non-daemon',
                target=non_daemon)
  n.daemon = False
  d.start()
  n.start()
  d.join(1)
  print 'd.is_alive()', d.is_alive()
  n.join()

poison pill 을 사용 하 는 것 이 좋 습 니 다.terminate()를 강제로 사용 한 후 join 을 사용 하여 상 태 를 업데이트 할 수 있 도록 하 는 것 이 좋 습 니 다.
프로 세 스 종료:

import multiprocessing
import time
def slow_worker():
  print 'Starting worker'
  time.sleep(0.1)
  print 'Finished worker'
if __name__ == '__main__':
  p = multiprocessing.Process(target=slow_worker)
  print 'BEFORE:', p, p.is_alive()
  p.start()
  print 'DURING:', p, p.is_alive()
  p.terminate()
  print 'TERMINATED:', p, p.is_alive()
  p.join()
  print 'JOINED:', p, p.is_alive()

①.==0 오류 가 발생 하지 않 음 
②.0 프로 세 스 에 오류 가 있 으 며 이 오류 코드 로 종료 합 니 다.
③.<0 프로 세 스 는 1-1*exitcode 신호 로 끝 납 니 다.
프로 세 스 종료 상태:

import multiprocessing
import sys
import time
def exit_error():
  sys.exit(1)
def exit_ok():
  return
def return_value():
  return 1
def raises():
  raise RuntimeError('There was an error!')
def terminated():
  time.sleep(3)
if __name__ == '__main__':
  jobs = []
  for f in [exit_error, exit_ok, return_value, raises, terminated]:
    print 'Starting process for', f.func_name
    j = multiprocessing.Process(target=f, name=f.func_name)
    jobs.append(j)
    j.start()
  jobs[-1].terminate()
  for j in jobs:
    j.join()
    print '%15s.exitcode = %s' % (j.name, j.exitcode)

편리 한 디 버 깅,logging 사용 가능
로그:

import multiprocessing
import logging
import sys
def worker():
  print 'Doing some work'
  sys.stdout.flush()
if __name__ == '__main__':
  multiprocessing.log_to_stderr()
  logger = multiprocessing.get_logger()
  logger.setLevel(logging.INFO)
  p = multiprocessing.Process(target=worker)
  p.start()
  p.join()

class 를 이용 하여 프로 세 스 를 만 들 고 하위 클래스 를 만 듭 니 다.
파생 프로 세 스:

import multiprocessing
class Worker(multiprocessing.Process):
  def run(self):
    print 'In %s' % self.name
    return
if __name__ == '__main__':
  jobs = []
  for i in range(5):
    p = Worker()
    jobs.append(p)
    p.start()
  for j in jobs:
    j.join()

python 프로 세 스 간 메시지 전달:

import multiprocessing
class MyFancyClass(object):
  def __init__(self, name):
    self.name = name
  def do_something(self):
    proc_name = multiprocessing.current_process().name
    print 'Doing something fancy in %s for %s!' % \
      (proc_name, self.name)
def worker(q):
  obj = q.get()
  obj.do_something()
if __name__ == '__main__':
  queue = multiprocessing.Queue()
  p = multiprocessing.Process(target=worker, args=(queue,))
  p.start()
  queue.put(MyFancyClass('Fancy Dan'))
  # Wait for the worker to finish
  queue.close()
  queue.join_thread()
  p.join()
import multiprocessing
import time
class Consumer(multiprocessing.Process):
  def __init__(self, task_queue, result_queue):
    multiprocessing.Process.__init__(self)
    self.task_queue = task_queue
    self.result_queue = result_queue
  def run(self):
    proc_name = self.name
    while True:
      next_task = self.task_queue.get()
      if next_task is None:
        # Poison pill means shutdown
        print '%s: Exiting' % proc_name
        self.task_queue.task_done()
        break
      print '%s: %s' % (proc_name, next_task)
      answer = next_task()
      self.task_queue.task_done()
      self.result_queue.put(answer)
    return
class Task(object):
  def __init__(self, a, b):
    self.a = a
    self.b = b
  def __call__(self):
    time.sleep(0.1) # pretend to take some time to do the work
    return '%s * %s = %s' % (self.a, self.b, self.a * self.b)
  def __str__(self):
    return '%s * %s' % (self.a, self.b)
if __name__ == '__main__':
  # Establish communication queues
  tasks = multiprocessing.JoinableQueue()
  results = multiprocessing.Queue()
  # Start consumers
  num_consumers = multiprocessing.cpu_count() * 2
  print 'Creating %d consumers' % num_consumers
  consumers = [ Consumer(tasks, results)
         for i in xrange(num_consumers) ]
  for w in consumers:
    w.start()
  # Enqueue jobs
  num_jobs = 10
  for i in xrange(num_jobs):
    tasks.put(Task(i, i))
  # Add a poison pill for each consumer
  for i in xrange(num_consumers):
    tasks.put(None)
  # Wait for all of the tasks to finish
  tasks.join()
  # Start printing results
  while num_jobs:
    result = results.get()
    print 'Result:', result
    num_jobs -= 1

Event 는 프로 세 스 간 에 상태 정 보 를 전달 할 수 있 는 간단 한 방법 을 제공 합 니 다.이 벤트 는 설정 과 설정 되 지 않 은 상 태 를 전환 할 수 있 습 니 다.선택 할 수 있 는 시간 초과 값 을 사용 하면 시간 대상 의 사용 자 는 상태 가 설정 되 지 않 은 것 을 기다 릴 수 있 습 니 다.
프로 세 스 간 신호 전달:

import multiprocessing
import time
def wait_for_event(e):
  """Wait for the event to be set before doing anything"""
  print 'wait_for_event: starting'
  e.wait()
  print 'wait_for_event: e.is_set()->', e.is_set()
def wait_for_event_timeout(e, t):
  """Wait t seconds and then timeout"""
  print 'wait_for_event_timeout: starting'
  e.wait(t)
  print 'wait_for_event_timeout: e.is_set()->', e.is_set()
if __name__ == '__main__':
  e = multiprocessing.Event()
  w1 = multiprocessing.Process(name='block', 
                 target=wait_for_event,
                 args=(e,))
  w1.start()
  w2 = multiprocessing.Process(name='nonblock', 
                 target=wait_for_event_timeout, 
                 args=(e, 2))
  w2.start()
  print 'main: waiting before calling Event.set()'
  time.sleep(3)
  e.set()
  print 'main: event is set'

Python 다 중 프로 세 스,일반적인 상황 은 Queue 가 전달 합 니 다.
Queue：

from multiprocessing import Process, Queue
def f(q):
  q.put([42, None, 'hello'])
if __name__ == '__main__':
  q = Queue()
  p = Process(target=f, args=(q,))
  p.start()
  print q.get()  # prints "[42, None, 'hello']"
  p.join()

다 중 스 레 드 우선 대기 열 큐:

import Queue
import threading
import time
exitFlag = 0
class myThread (threading.Thread):
  def __init__(self, threadID, name, q):
    threading.Thread.__init__(self)
    self.threadID = threadID
    self.name = name
    self.q = q
  def run(self):
    print "Starting " + self.name
    process_data(self.name, self.q)
    print "Exiting " + self.name
def process_data(threadName, q):
  while not exitFlag:
    queueLock.acquire()
    if not workQueue.empty():
      data = q.get()
      queueLock.release()
      print "%s processing %s" % (threadName, data)
    else:
      queueLock.release()
    time.sleep(1)
threadList = ["Thread-1", "Thread-2", "Thread-3"]
nameList = ["One", "Two", "Three", "Four", "Five"]
queueLock = threading.Lock()
workQueue = Queue.Queue(10)
threads = []
threadID = 1
# Create new threads
for tName in threadList:
  thread = myThread(threadID, tName, workQueue)
  thread.start()
  threads.append(thread)
  threadID += 1
# Fill the queue
queueLock.acquire()
for word in nameList:
  workQueue.put(word)
queueLock.release()
# Wait for queue to empty
while not workQueue.empty():
  pass
# Notify threads it's time to exit
exitFlag = 1
# Wait for all threads to complete
for t in threads:
  t.join()
print "Exiting Main Thread"

다 중 프로 세 스 가 Queue 통신 을 사용 하 는 예

import time
from multiprocessing import Process,Queue
MSG_QUEUE = Queue(5)
def startA(msgQueue):
  while True:
    if msgQueue.empty() > 0:
      print ('queue is empty %d' % (msgQueue.qsize()))
    else:
      msg = msgQueue.get()
      print( 'get msg %s' % (msg,))
    time.sleep(1)
def startB(msgQueue):
  while True:
    msgQueue.put('hello world')
    print( 'put hello world queue size is %d' % (msgQueue.qsize(),))
    time.sleep(3)
if __name__ == '__main__':
  processA = Process(target=startA,args=(MSG_QUEUE,))
  processB = Process(target=startB,args=(MSG_QUEUE,))
  processA.start()
  print( 'processA start..')

주 프로 세 스 는 Queue 형식의 변 수 를 정의 하고 process 의 args 매개 변수 로 하위 프로 세 스 processA 와 processB 에 전달 합 니 다.두 프로 세 스 는 하나의 대기 열 에 데 이 터 를 쓰 고 하나의 데 이 터 를 읽 습 니 다.
파 이 썬 관련 내용 에 관심 이 있 는 독자 들 은 본 사이트 의 주 제 를 볼 수 있다.
본 논문 에서 말 한 것 이 여러분 의 Python 프로 그래 밍 에 도움 이 되 기 를 바 랍 니 다.