Python多进程multiprocessing用法实例分析

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

import multiprocessingimport timedef 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()

import multiprocessingimport timeimport sysdef daemon():  name = multiprocessing.current_process().name  print 'Starting:', name  time.sleep(2)  print 'Exiting :', namedef non_daemon():  name = multiprocessing.current_process().name  print 'Starting:', name  print 'Exiting :', nameif __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()

import multiprocessingimport timedef 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()

import multiprocessingimport sysimport timedef exit_error():  sys.exit(1)def exit_ok():  returndef return_value():  return 1def 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)

import multiprocessingimport loggingimport sysdef 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()

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

import multiprocessingclass 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 multiprocessingimport timeclass 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)    returnclass 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

import multiprocessingimport timedef 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'

from multiprocessing import Process, Queuedef 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 Queueimport threadingimport timeexitFlag = 0class 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.namedef 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 threadsfor tName in threadList:  thread = myThread(threadID, tName, workQueue)  thread.start()  threads.append(thread)  threadID += 1# Fill the queuequeueLock.acquire()for word in nameList:  workQueue.put(word)queueLock.release()# Wait for queue to emptywhile not workQueue.empty():  pass# Notify threads it's time to exitexitFlag = 1# Wait for all threads to completefor t in threads:  t.join()print "Exiting Main Thread"

import timefrom multiprocessing import Process,QueueMSG_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..')