python之进程

Posted 2021-02-01 yhw-miracle

（首发于 2018 年 7 月 31 日）

进程（Process）是计算机中的程序关于某数据集合上的一次运行活动，是系统进行资源分配和调度的基本单位，是操作系统结构的基础。为了提高程序运行的效率，python 中提供了多进程机制。 python 实现多进程的方式主要有两种，一种是使用 os 模块中的 fork 方法，另一种是使用 multiprocessing 模块。这两种方法的区别在于前者适用于 Unix/Linux 操作系统，对 Windows 系统不支持，后者则是跨平台的实现方式。

1. 使用 os 模块中的 fork 方法实现多进程

python 的 os 模块封装了常见的系统调用，其中就有 fork 方法。fork 方法来自于 Unix/Linux 操作系统中提供的一个 fork 系统调用，这个方法非常特殊。普通方法都是调用一次，返回一次。而fork 方法是调用一次，返回两次，原因在操作系统将当前系统（父进程）复制出一份进程（子进程），这两个进程几乎完全相同，于是 fork 方法分别在父进程和子进程中返回。子进程中永远返回 0，父进程中返回的是子进程的 ID。具体情况，见如下代码，其中，os 模块中的 getpid 方法用于返回当前进程的 ID，getppid 方法用于获取父进程的 ID。

 1 import os
 2 if __name__ == ‘__main__‘:
 3     print ‘current Process (%s) start ...‘ % (os.getpid())
 4     pid = os.fork()
 5     if pid < 0:
 6         print ‘error in fork.‘
 7     elif pid > 0:
 8         print ‘I (%s) create a child process (%s)‘ % (os.getpid(), pid)
 9     else:
10         print ‘I am child process (%s) and my parent process is (%s)‘ % (os.getpid(), os.getppid())

2. 使用 multiprocessing 模块创建多进程

multiprocessing 模块提供了一个 Process 类来描述进程对象。创建子进程时，只需要传入一个执行函数和函数参数，即可完成一个 Process 实例的创建，用 start() 方法启动进程，用 join() 实现进程间的同步。下面，用一个例子来演示说明。

 1 import os
 2 from multiprocessing import Process
 3 def run_process(name):
 4     print ‘Child process %s (%s) Running...‘ % (name, os.getpid())
 5 
 6 
 7 if __name__ == ‘__main__‘:
 8     print ‘Parent process %s.‘ % os.getpid()
 9     for i in range(5):
10         p = Process(target=run_process, args=(str(i),))
11         print ‘Process will start.‘
12         p.start()
13     p.join()
14     print ‘Process end.‘

进程具有并发性，共享性，独立性和异步性，以上代码中子进程输出语句说明了进程的异步性。

3. multiprocessing 模块提供了 Pool 类来代表进程池对象

以上两种创建进程的方法在启动大量子进程时，需要手动限制进程数量，显得太过繁琐，而使用进程池批量创建子进程的方式更加方便。

multiprocessing 模块提供了 Pool 类来代表进程池对象，Pool 类可以提供制定数量的进程供用户调用，默认大小是 CPU 的核数。当有新的请求提交到 Pool 中时，如果进程池还没有满，那么就会创建一个新的进程用来执行该请求；如果进程池中的进程数已经达到规定的最大值，那么该请求就好等待，直到池中有进程结束，才会创建新的进程来处理它。下面，用代码来演示进程池对象。

 1 from multiprocessing import Pool
 2 import os, time, random
 3 
 4 
 5 def run_task(name):
 6     print ‘Task %s (pid = %s) is running...‘ % (name, os.getpid())
 7     time.sleep(random.random() * 3)
 8     print ‘Task %s end.‘ % name
 9 
10 
11 if __name__ == ‘__main__‘:
12     print ‘current process %s.‘ % os.getpid()
13     p = Pool(processes=3)
14     for i in range(5):
15         p.apply_async(run_task, args=(i,))
16     print ‘Waiting for all subprocess done...‘
17     p.close()
18     p.join()
19     print ‘All subprocess done.‘

注意： Pool 对象调用 join() 方法会等待所有子进程执行完毕，调用 join() 方法之前必须调用 close() 方法，调用 close() 方法之后就不能继续添加新的进程了。

4. 进程间通信

操作系统中存在大量进程，那么进程间的通信是必不可少的。python 提供了 Queue 和 Pipe 两张方式实现进程间的通信，两张区别在于，Pipe 常用来在两个进程间通信，Queue 用来在多个进程间实现通信。

4.1 Queue 实现进程通信

Queue 是多进程安全的队列，可以使用 Queue 实现多进程间的数据传递，有两个方法 Put 和 Get 可以进行 Queue 操作。

Put 方法用以插入数据到队列中，它还有两个可选数据：blocked 和 timeout。如果 blocked 为 True （默认值），并且 timeout 为正值，该方法会阻塞 timeout 指定的时间，直到该队列有剩余空间；如果超时，会抛出 Queue.Full 异常。如果 blocked 为 False，但该队列以满，会立即抛出 Queue.Full 异常。

Get 方法可以从队列读取并且删除一个元素。Get 方法也有两个可选参数：blocked 和 timeout。如果 blocked 为 True （默认值），并且 timeout 为正值，那么在等待时间内没有取到任何元素，会抛出 Queue.Empty 异常；如果 blocked 为 False，有两张情况：如果队列有元素可以用，则立即会返回该值；否则，如果队列为空，立即抛出 Queue.Empty 异常。

下面用代码来演示 Queue 实现进程间通信。

 1 from multiprocessing import Process, Queue
 2 import os, time, random
 3 
 4 
 5 def process_write(q, urls):
 6     print(‘Process (%s) is writing...‘ % os.getpid())
 7     for url in urls:
 8         q.put(url)
 9         print(‘Put %s to queue...‘ % url)
10         time.sleep(random.random())
11 
12 
13 def process_read(q):
14     print(‘Process (%s) is reading...‘ % os.getpid())
15     while True:
16         url = q.get(True)
17         print(‘Get %s from queue...‘ % url)
18 
19 
20 if __name__ == ‘__main__‘:
21     q = Queue()
22     process_write1 = Process(target=process_write, args=(q, [‘url1‘, ‘url2‘, ‘url3‘]))
23     process_write2 = Process(target=process_write, args=(q, [‘url4‘, ‘url5‘, ‘url6‘]))
24     process_reader = Process(target=process_read, args=(q,))
25 
26     process_write1.start()
27     process_write2.start()
28     process_reader.start()
29 
30     process_write1.join()
31     process_write2.join()
32 
33     process_reader.terminate()

4.2 Pipe 实现进程通信

Pipe 通常用来在两个进程间进行通信，两个进程分别位于管道的两端。Pipe 方法返回 (conn1, conn2) 代表管道的两端，该方法还有 duplex 参数，如果 duplex 参数为 True （默认值），那么这个管道是全双工模式，也就是管道两端均可以同时接收和返送数据；如果 duplex 参数为 False，conn1 只负责接收消息，conn2 只负责发送消息。

send 和 recv 方法分别是发送和接收消息的方法，如果没有消息可接收，recv 方法会一直阻塞，如果管道关闭，那么 recv 方法会抛出 EOFError。下面通过代码来演示 Pipe 实现进程通信。

 1 import multiprocessing, random, os, time
 2 
 3 
 4 def process_send(pipe, urls):
 5     for url in urls:
 6         print(‘Process (%s) send: %s‘ % (os.getpid(), url))
 7         pipe.send(url)
 8         time.sleep(random.random())
 9 
10 
11 def process_recv(pipe):
12     while True:
13         print(‘Process (%s) recv: %s‘ % (os.getpid(), pipe.recv()))
14         time.sleep(random.random())
15 
16 
17 if __name__ == ‘__main__‘:
18     pipe = multiprocessing.Pipe()
19     p1 = multiprocessing.Process(target=process_send, args=(pipe[0], [‘url‘ + str(i) for i in range(10)]))
20     p2 = multiprocessing.Process(target=process_recv, args=(pipe[1],))
21 
22     p1.start()
23     p2.start()
24 
25     p1.join()
26     p2.terminate()