yhw-miracle

python之进程

痛点就是起点 writed in

本文为痛点就是起点原创文章,可以随意转载,但需注明出处。

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

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

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

import os
if __name__ == '__main__':
    print 'current Process (%s) start ...' % (os.getpid())
    pid = os.fork()
    if pid < 0:
        print 'error in fork.'
    elif pid > 0:
        print 'I (%s) create a child process (%s)' % (os.getpid(), pid)
    else:
        print 'I am child process (%s) and my parent process is (%s)' % (os.getpid(), os.getppid())

使用 multiprocessing 模块创建多进程

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

import os
from multiprocessing import Process
def run_process(name):
    print 'Child process %s (%s) Running...' % (name, os.getpid())


if __name__ == '__main__':
    print 'Parent process %s.' % os.getpid()
    for i in range(5):
        p = Process(target=run_process, args=(str(i),))
        print 'Process will start.'
        p.start()
    p.join()
    print 'Process end.'

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

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

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

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

from multiprocessing import Pool
import os, time, random


def run_task(name):
    print 'Task %s (pid = %s) is running...' % (name, os.getpid())
    time.sleep(random.random() * 3)
    print 'Task %s end.' % name


if __name__ == '__main__':
    print 'current process %s.' % os.getpid()
    p = Pool(processes=3)
    for i in range(5):
        p.apply_async(run_task, args=(i,))
    print 'Waiting for all subprocess done...'
    p.close()
    p.join()
    print 'All subprocess done.'

进程间通信

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

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 实现进程间通信。

from multiprocessing import Process, Queue
import os, time, random


def process_write(q, urls):
    print('Process (%s) is writing...' % os.getpid())
    for url in urls:
        q.put(url)
        print('Put %s to queue...' % url)
        time.sleep(random.random())


def process_read(q):
    print('Process (%s) is reading...' % os.getpid())
    while True:
        url = q.get(True)
        print('Get %s from queue...' % url)


if __name__ == '__main__':
    q = Queue()
    process_write1 = Process(target=process_write, args=(q, ['url1', 'url2', 'url3']))
    process_write2 = Process(target=process_write, args=(q, ['url4', 'url5', 'url6']))
    process_reader = Process(target=process_read, args=(q,))

    process_write1.start()
    process_write2.start()
    process_reader.start()

    process_write1.join()
    process_write2.join()

    process_reader.terminate()

Pipe 实现进程通信

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

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

import multiprocessing, random, os, time


def process_send(pipe, urls):
    for url in urls:
        print('Process (%s) send: %s' % (os.getpid(), url))
        pipe.send(url)
        time.sleep(random.random())


def process_recv(pipe):
    while True:
        print('Process (%s) recv: %s' % (os.getpid(), pipe.recv()))
        time.sleep(random.random())


if __name__ == '__main__':
    pipe = multiprocessing.Pipe()
    p1 = multiprocessing.Process(target=process_send, args=(pipe[0], ['url' + str(i) for i in range(10)]))
    p2 = multiprocessing.Process(target=process_recv, args=(pipe[1],))

    p1.start()
    p2.start()

    p1.join()
    p2.terminate()

python 进程
知识总结

欢迎关注,我们一起进行认知迭代!


痛点就是起点

© 2016 - 2020 基于 jekyll | Github Pags | iconfont By yhw-miracle