Python multiprocessing多进程原理与应用示例

多进程基础原理

在Python中，multiprocessing包提供了跨平台的多进程管理功能。与多线程不同，多进程为每个进程分配了独立的内存空间，因此不存在全局解释器锁（GIL）争抢的问题。在多核CPU环境下，使用多进程可以充分利用硬件资源，显著提高程序的运行效率。

Process对象与生命周期

Process类是创建子进程的核心对象。其构造函数接受以下参数：

• group：通常不使用，保留为None。
• target：指定子进程要执行的函数。
• name：定义进程的名称，便于调试和识别。
• args：以元组形式传递给目标函数的位置参数。
• kwargs：以字典形式传递给目标函数的关键字参数。

进程的生命周期由几个关键方法控制：

1. start()：启动子进程，使其开始执行目标函数。
2. join(timeout=None)：阻塞主进程，直到子进程结束。可以设置超时时间，超时后主进程继续执行。
3. terminate()：强制终止子进程。调用后通常需要紧跟join()来更新进程状态。
4. is_alive()：检查进程是否仍在运行。

此外，daemon属性决定了子进程是否随主进程结束而终止。若设为True，子进程将随主进程退出而强制结束；若为False（默认值），子进程会独立运行直至完成。注意，daemon必须在调用start()之前设置。

进程状态与退出码

进程结束后，可以通过exitcode属性判断其退出状态：

• 0：正常退出，无错误。
• > 0：进程发生错误，以该退出码结束。
• < 0：进程被信号终止。

进程池Pool的使用

当需要管理大量子进程时，手动创建Process对象效率较低。此时可以使用Pool（进程池）来自动管理进程的生命周期和任务分配。

创建与任务提交

通过Pool()构造函数可以创建一个进程池，默认情况下，进程池中的进程数量等于机器的CPU核心数（可通过multiprocessing.cpu_count()获取）。当提交的任务数超过进程池容量时，进程池会自动复用已完成的进程来执行新任务。

使用apply_async()方法可以异步提交任务，该方法返回一个AsyncResult对象，可用于获取任务结果或检查状态。

进程池的生命周期管理

进程池的使用遵循严格的顺序：

1. 提交所有任务。
2. 调用p.close()：关闭进程池，禁止再提交新任务。
3. 调用p.join()：阻塞主进程，等待所有子进程执行完毕。

注意，join()必须在close()之后调用。

进程间通信与共享

由于多进程拥有独立的内存空间，父进程与子进程之间无法直接共享普通变量。Python提供了多种机制来实现进程间通信（IPC）和共享内存。

Queue消息队列

multiprocessing.Queue是进程间通信的常用方式。它支持多线程和多进程安全地传递数据。

• put(value, block=True, timeout=None)：向队列添加值。可设置超时时间。
• get(block=True, timeout=None)：从队列获取并移除值。可设置超时时间。
• empty()：判断队列是否为空。
• full()：判断队列是否已满。
• qsize()：返回队列中消息的大致数量。

在进程池Pool中使用Queue时，必须使用Manager().Queue()，因为普通的Queue无法在进程池的子进程间正确共享。

共享内存对象

对于需要频繁读写的大型数据结构，可以使用共享内存对象来避免数据复制的开销。

• multiprocessing.Value(typecode_or_type, value, lock=True)：创建一个共享的基本类型变量（如整数、浮点数）。
• multiprocessing.Array(typecode_or_type, sequence, lock=True)：创建一个共享的数组。

通过Value和Array，子进程可以直接修改这些对象，修改结果对父进程和其他子进程可见。

代码示例

1. 基础Process使用

import multiprocessing
import time
import cv2

def daemon1(image):
  name = multiprocessing.current_process().name
  for i in range(50):
    image = cv2.GaussianBlur(image, (3, 3), 1)
    time.sleep(0.1)
  print 'daemon1 done!'
  cv2.imshow('daemon1', image)

def daemon2(image):
  name = multiprocessing.current_process().name
  for i in range(50):
    image = cv2.GaussianBlur(image, (3, 3), 1)
    time.sleep(0.5)
  print 'daemon2 done!'
  cv2.imshow('daemon2', image)

if __name__ == '__main__':
  t1 = time.time()
  number_kernel = multiprocessing.cpu_count()
  print 'We have {0} kernels'.format(number_kernel)
  p1 = multiprocessing.Process(name='daemon1',
                target=daemon1,args= (cv2.imread('./p1.jpg'),))
  p1.daemon = False
  p2 = multiprocessing.Process(name='daemon2',
                target=daemon2, args=(cv2.imread('./p2.jpg'),))
  p2.daemon = False
  p1.start()
  p2.start()
  print 'p1 is {0}'.format(p1.is_alive())
  p1.terminate()
  p1.join()
  print 'p1 is {0}'.format(p1.is_alive())
  print 'p2 is {0}'.format(p2.is_alive())
  p2.join()
  t2 = time.time()
  print '!!!!!!!!!!!!!!!!!!!!OK!!!!!!!!!!!!!!!!!!!!!'
  print 'total time is {0}'.format(t2-t1)
  print 'p1.exitcode = {0}'.format(p1.exitcode)
  print 'p2.exitcode = {0}'.format(p2.exitcode)

2. 进程池Pool使用

from multiprocessing import Pool
import os, time
def long_time_task(name):
  print 'Run task {0} ({1})'.format(name,os.getpid())
  start = time.time()
  time.sleep(3)
  end = time.time()
  print 'Task {0} runs {1:.2f} seconds.'.format(name,end - start)
if __name__=='__main__':
  print 'Parent process ({0})'.format(os.getpid)
  p = Pool()
  for i in range(12):
    p.apply_async(long_time_task, args=(i,))
  print 'Waiting for all subprocesses done...'
  p.close()
  p.join()
  print 'All subprocesses done.'

3. Queue消息队列通信

from multiprocessing import Process, Queue
import os, time, random
def write(q):
  for value in ['A', 'B', 'C']:
    print 'Put {0} to queue...'.format(value)
    q.put(value)
    time.sleep(random.random())
def read(q):
  while True:
    if not q.empty():
      value = q.get(True)
      print 'Get {0} from queue.'.format(value)
      time.sleep(random.random())
    else:
      break
if __name__=='__main__':
  q = multiprocessing.Queue()
  pw = Process(target=write, args=(q,))
  pr = Process(target=read, args=(q,))
  pw.start()
  pw.join()
  pr.start()
  pr.join()

4. 进程池中使用Queue

from multiprocessing import Process, Queue
import os, time, random
def write(q):
  for value in ['A', 'B', 'C']:
    print 'Put {0} to queue...'.format(value)
    q.put(value)
    time.sleep(random.random())
def read(q):
  while True:
    if not q.empty():
      value = q.get(True)
      print 'Get {0} from queue.'.format(value)
      time.sleep(random.random())
    else:
      break
if __name__=='__main__':
  manager = multiprocessing.Manager()
  q = manager.Queue()
  p = Pool()
  pw = p.apply_async(write, args=(q,))
  time.sleep(2)
  pr = p.apply_async(read, args=(q,))
  p.close()
  p.join()
  if not q.empty():
    print 'q is not empty...'
  else:
    print 'q is empty...'
  print 'OK'
  if not q.empty():
    print 'q is not empty...'
  else:
    print 'q is empty...'
  print 'done...'

5. 普通变量与共享内存对比

以下示例展示了普通变量在子进程中修改后不会影响父进程，而使用Value和Array可以实现共享内存修改。

# 普通变量（不共享）
import multiprocessing
from multiprocessing import Pool
def changevalue(n, a):
  n = 3.14
  a[0] = 5
if __name__ == '__main__':
  num = 0
  arr = range(10)
  p = Pool()
  p1 = p.apply_async(changevalue, args=(num, arr))
  p.close()
  p.join()
  print num
  print arr[:]

# 共享内存（共享）
import multiprocessing
def changevalue(n, a):
  n.value = 3.14
  a[0] = 5
if __name__ == '__main__':
  num = multiprocessing.Value('d', 0.0)
  arr = multiprocessing.Array('i', range(10))
  p = multiprocessing.Process(target=changevalue, args=(num, arr))
  p.start()
  p.join()
  print num.value
  print arr[:]

# 进程池中使用共享内存
import multiprocessing
from multiprocessing import Pool
def changevalue(n, a):
  n.value = 3.14
  a[0] = 5
if __name__ == '__main__':
  num = multiprocessing.Value('d', 0.0)
  arr = multiprocessing.Array('i', range(10))
  p = Pool()
  p1 = p.apply_async(changevalue, args=(num, arr))
  p.close()
  p.join()
  print num.value
  print arr[:]

以上就是Python multiprocessing多进程原理与应用示例的详细内容，更多关于Python多进程编程的资料请关注本站其它相关文章！