多进程编程知识是Python程序员进阶高级的必备知识点,咱们平时习惯了使用multiprocessing库来操纵多进程,可是并不知道它的具体实现原理。下面我对多进程的经常使用知识点都简单列了一遍,使用原生的多进程方法调用,帮助读者理解多进程的实现机制。代码跑在linux环境下。没有linux条件的,可使用docker或者虚拟机运行进行体验。python
docker pull python:2.7
Python生成子进程使用os.fork()
,它将产生一个子进程。fork调用同时在父进程和主进程同时返回,在父进程中返回子进程的pid,在子进程中返回0,若是返回值小于零,说明子进程产生失败,通常是由于操做系统资源不足。linux
import os def create_child(): pid = os.fork() if pid > 0: print 'in father process' return True elif pid == 0: print 'in child process' return False else: raise
咱们调用create_child
方法屡次就能够生成多个子进程,前提是必须保证create_child
是在父进程里执行,若是是子进程,就不要在调用了。 程序员
# coding: utf-8 # child.py import os def create_child(i): pid = os.fork() if pid > 0: print 'in father process' return pid elif pid == 0: print 'in child process', i return 0 else: raise for i in range(10): # 循环10次,建立10个子进程 pid = create_child(i) # pid==0是子进程,应该当即退出循环,不然子进程也会继续生成子进程 # 子子孙孙,那就生成太多进程了 if pid == 0: break
运行python child.py
,输出redis
in father process in father process in child process 0 in child process 1 in father process in child process 2 in father process in father process in child process 3 in father process in child process 4 in child process 5 in father process in father process in child process 6 in child process 7 in father process in child process 8 in father process in child process 9
使用time.sleep可使进程休眠任意时间,单位为秒,能够是小数docker
import time for i in range(5): print 'hello' time.sleep(1) # 睡1s
使用os.kill(pid, sig_num)能够向进程号为pid的子进程发送信号,sig_num经常使用的有SIGKILL(暴力杀死,至关于kill -9),SIGTERM(通知对方退出,至关于kill不带参数),SIGINT(至关于键盘的ctrl+c)。macos
# coding: utf-8 # kill.py import os import time import signal def create_child(): pid = os.fork() if pid > 0: return pid elif pid == 0: return 0 else: raise pid = create_child() if pid == 0: while True: # 子进程死循环打印字符串 print 'in child process' time.sleep(1) else: print 'in father process' time.sleep(5) # 父进程休眠5s再杀死子进程 os.kill(pid, signal.SIGKILL) time.sleep(5) # 父进程继续休眠5s观察子进程是否还有输出
运行python kill.py
,咱们看到控制台输出以下编程
in father process in child process # 等1s in child process # 等1s in child process # 等1s in child process # 等1s in child process # 等了5s
说明os.kill执行以后,子进程已经中止输出了api
在上面的例子中,os.kill执行完以后,咱们经过ps -ef|grep python快速观察进程的状态,能够发现子进程有一个奇怪的显示<defunct>
网络
root 12 1 0 11:22 pts/0 00:00:00 python kill.py root 13 12 0 11:22 pts/0 00:00:00 [python] <defunct>
待父进程终止后,子进程也一块消失了。那<defunct>
是什么含义呢? 它的含义是「僵尸进程」。子进程结束后,会当即成为僵尸进程,僵尸进程占用的操做系统资源并不会当即释放,它就像一具尸体啥事也不干,可是仍是持续占据着操做系统的资源(内存等)。并发
若是完全干掉僵尸进程?父进程须要调用waitpid(pid, options)函数,「收割」子进程,这样子进程才能够灰飞烟灭。waitpid函数会返回子进程的退出状态,它就像子进程留下的临终遗言,必须等父进程听到后才能完全瞑目。
# coding: utf-8 import os import time import signal def create_child(): pid = os.fork() if pid > 0: return pid elif pid == 0: return 0 else: raise pid = create_child() if pid == 0: while True: # 子进程死循环打印字符串 print 'in child process' time.sleep(1) else: print 'in father process' time.sleep(5) # 父进程休眠5s再杀死子进程 os.kill(pid, signal.SIGTERM) ret = os.waitpid(pid, 0) # 收割子进程 print ret # 看看到底返回了什么 time.sleep(5) # 父进程继续休眠5s观察子进程是否还存在
运行python kill.py输出以下
in father process in child process in child process in child process in child process in child process in child process (125, 9)
咱们看到waitpid返回了一个tuple,第一个是子进程的pid,第二个9是什么含义呢,它在不一样的操做系统上含义不尽相同,不过在Unix上,它一般的value是一个16位的整数值,前8位表示进程的退出状态,后8位表示致使进程退出的信号的整数值。因此本例中退出状态位0,信号编号位9,还记得kill -9
这个命令么,就是这个9表示暴力杀死进程。
若是咱们将os.kill换一个信号才看结果,好比换成os.kill(pid, signal.SIGTERM),能够看到返回结果变成了(138, 15)
,15就是SIGTERM信号的整数值。
waitpid(pid, 0)
还能够起到等待子进程结束的功能,若是子进程不结束,那么该调用会一直卡住。
SIGTERM信号默认处理动做就是退出进程,其实咱们还能够设置SIGTERM信号的处理函数,使得它不退出。
# coding: utf-8 import os import time import signal def create_child(): pid = os.fork() if pid > 0: return pid elif pid == 0: return 0 else: raise pid = create_child() if pid == 0: signal.signal(signal.SIGTERM, signal.SIG_IGN) while True: # 子进程死循环打印字符串 print 'in child process' time.sleep(1) else: print 'in father process' time.sleep(5) # 父进程休眠5s再杀死子进程 os.kill(pid, signal.SIGTERM) # 发一个SIGTERM信号 time.sleep(5) # 父进程继续休眠5s观察子进程是否还存在 os.kill(pid, signal.SIGKILL) # 发一个SIGKILL信号 time.sleep(5) # 父进程继续休眠5s观察子进程是否还存在
咱们在子进程里设置了信号处理函数,SIG_IGN表示忽略信号。咱们发现第一次调用os.kill以后,子进程会继续输出。说明子进程没有被杀死。第二次os.kill以后,子进程终于中止了输出。
接下来咱们换一个自定义信号处理函数,子进程收到SIGTERM以后,打印一句话再退出。
# coding: utf-8 import os import sys import time import signal def create_child(): pid = os.fork() if pid > 0: return pid elif pid == 0: return 0 else: raise def i_will_die(sig_num, frame): # 自定义信号处理函数 print "child will die" sys.exit(0) pid = create_child() if pid == 0: signal.signal(signal.SIGTERM, i_will_die) while True: # 子进程死循环打印字符串 print 'in child process' time.sleep(1) else: print 'in father process' time.sleep(5) # 父进程休眠5s再杀死子进程 os.kill(pid, signal.SIGTERM) time.sleep(5) # 父进程继续休眠5s观察子进程是否还存在
输出以下
in father process in child process in child process in child process in child process in child process child will die
信号处理函数有两个参数,第一个sig_num表示被捕获信号的整数值,第二个frame不太好理解,通常也不多用。它表示被信号打断时,Python的运行的栈帧对象信息。读者能够没必要深度理解。
下面咱们使用多进程进行一个计算圆周率PI。对于圆周率PI有一个数学极限公式,咱们将使用该公司来计算圆周率PI。
先使用单进程版本
import math def pi(n): s = 0.0 for i in range(n): s += 1.0/(2*i+1)/(2*i+1) return math.sqrt(8 * s) print pi(10000000)
输出
3.14159262176
这个程序跑了有一小会才出结果,不过这个值已经很是接近圆周率了。
接下来咱们用多进程版本,咱们用redis进行进程间通讯。
# coding: utf-8 import os import sys import math import redis def slice(mink, maxk): s = 0.0 for k in range(mink, maxk): s += 1.0/(2*k+1)/(2*k+1) return s def pi(n): pids = [] unit = n / 10 client = redis.StrictRedis() client.delete("result") # 保证结果集是干净的 del client # 关闭链接 for i in range(10): # 分10个子进程 mink = unit * i maxk = mink + unit pid = os.fork() if pid > 0: pids.append(pid) else: s = slice(mink, maxk) # 子进程开始计算 client = redis.StrictRedis() client.rpush("result", str(s)) # 传递子进程结果 sys.exit(0) # 子进程结束 for pid in pids: os.waitpid(pid, 0) # 等待子进程结束 sum = 0 client = redis.StrictRedis() for s in client.lrange("result", 0, -1): sum += float(s) # 收集子进程计算结果 return math.sqrt(sum * 8) print pi(10000000)
咱们将级数之和的计算拆分红10个子进程计算,每一个子进程负责1/10的计算量,并将计算的中间结果扔到redis的队列中,而后父进程等待全部子进程结束,再将队列中的数据所有汇总起来计算最终结果。
输出以下
3.14159262176
这个结果和单进程结果一致,可是花费的时间要缩短了很多。
这里咱们之因此使用redis做为进程间通讯方式,是由于进程间通讯是一个比较复杂的技术,接下来咱们将会使用进程间通讯技术来替换掉这里的redis。
使用文件进行通讯是最简单的一种通讯方式,子进程将结果输出到临时文件,父进程从文件中读出来。文件名使用子进程的进程id来命名。进程随时均可以经过os.getpid()
来获取本身的进程id。
# coding: utf-8 import os import sys import math def slice(mink, maxk): s = 0.0 for k in range(mink, maxk): s += 1.0/(2*k+1)/(2*k+1) return s def pi(n): pids = [] unit = n / 10 for i in range(10): # 分10个子进程 mink = unit * i maxk = mink + unit pid = os.fork() if pid > 0: pids.append(pid) else: s = slice(mink, maxk) # 子进程开始计算 with open("%d" % os.getpid(), "w") as f: f.write(str(s)) sys.exit(0) # 子进程结束 sums = [] for pid in pids: os.waitpid(pid, 0) # 等待子进程结束 with open("%d" % pid, "r") as f: sums.append(float(f.read())) os.remove("%d" % pid) # 删除通讯的文件 return math.sqrt(sum(sums) * 8) print pi(10000000)
输出
3.14159262176
管道是Unix进程间通讯最经常使用的方法之一,它经过在父子进程之间开通读写通道来进行双工交流。咱们经过os.read()和os.write()来对文件描述符进行读写操做,使用os.close()关闭描述符。
上图为单进程的管道
上图为父子进程分离后的管道
# coding: utf-8 import os import sys import math def slice(mink, maxk): s = 0.0 for k in range(mink, maxk): s += 1.0/(2*k+1)/(2*k+1) return s def pi(n): childs = {} unit = n / 10 for i in range(10): # 分10个子进程 mink = unit * i maxk = mink + unit r, w = os.pipe() pid = os.fork() if pid > 0: childs[pid] = r # 将子进程的pid和读描述符存起来 os.close(w) # 父进程关闭写描述符,只读 else: os.close(r) # 子进程关闭读描述符,只写 s = slice(mink, maxk) # 子进程开始计算 os.write(w, str(s)) os.close(w) # 写完了,关闭写描述符 sys.exit(0) # 子进程结束 sums = [] for pid, r in childs.items(): sums.append(float(os.read(r, 1024))) os.close(r) # 读完了,关闭读描述符 os.waitpid(pid, 0) # 等待子进程结束 return math.sqrt(sum(sums) * 8) print pi(10000000)
输出
3.14159262176
套接字无疑是通讯使用最为普遍的方式了,它不但能跨进程还能跨网络。今天英特网能发达成这样,全拜套接字所赐。不过做为同一个机器的多进程通讯仍是挺浪费的。暂不讨论这个,仍是先看看它如何使用吧。
# coding: utf-8 import os import sys import math import socket def slice(mink, maxk): s = 0.0 for k in range(mink, maxk): s += 1.0/(2*k+1)/(2*k+1) return s def pi(n): childs = [] unit = n / 10 servsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 注意这里的AF_INET表示普通套接字 servsock.bind(("localhost", 0)) # 0表示随机端口 server_address = servsock.getsockname() # 拿到随机出来的地址,给后面的子进程使用 servsock.listen(10) # 监听子进程链接请求 for i in range(10): # 分10个子进程 mink = unit * i maxk = mink + unit pid = os.fork() if pid > 0: childs.append(pid) else: servsock.close() # 子进程要关闭servsock引用 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.connect(server_address) # 链接父进程套接字 s = slice(mink, maxk) # 子进程开始计算 sock.sendall(str(s)) sock.close() # 关闭链接 sys.exit(0) # 子进程结束 sums = [] for pid in childs: conn, _ = servsock.accept() # 接收子进程链接 sums.append(float(conn.recv(1024))) conn.close() # 关闭链接 for pid in childs: os.waitpid(pid, 0) # 等待子进程结束 servsock.close() # 关闭套接字 return math.sqrt(sum(sums) * 8) print pi(10000000)
输出
3.14159262176
当同一个机器的多个进程使用普通套接字进行通讯时,须要通过网络协议栈,这很是浪费,由于同一个机器根本没有必要走网络。因此Unix提供了一个套接字的特殊版本,它使用和套接字一摸同样的api,可是地址再也不是网络端口,而是文件。至关于咱们经过某个特殊文件来进行套接字通讯。
# coding: utf-8 import os import sys import math import socket def slice(mink, maxk): s = 0.0 for k in range(mink, maxk): s += 1.0/(2*k+1)/(2*k+1) return s def pi(n): server_address = "/tmp/pi_sock" # 套接字对应的文件名 childs = [] unit = n / 10 servsock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) # 注意AF_UNIX表示「域套接字」 servsock.bind(server_address) servsock.listen(10) # 监听子进程链接请求 for i in range(10): # 分10个子进程 mink = unit * i maxk = mink + unit pid = os.fork() if pid > 0: childs.append(pid) else: servsock.close() # 子进程要关闭servsock引用 sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) sock.connect(server_address) # 链接父进程套接字 s = slice(mink, maxk) # 子进程开始计算 sock.sendall(str(s)) sock.close() # 关闭链接 sys.exit(0) # 子进程结束 sums = [] for pid in childs: conn, _ = servsock.accept() # 接收子进程链接 sums.append(float(conn.recv(1024))) conn.close() # 关闭链接 for pid in childs: os.waitpid(pid, 0) # 等待子进程结束 servsock.close() # 关闭套接字 os.unlink(server_address) # 移除套接字文件 return math.sqrt(sum(sums) * 8) print pi(10000000)
输出
3.14159262176
咱们知道跨网络通讯免不了要经过套接字进行通讯,可是本例的多进程是在同一个机器上,用不着跨网络,使用普通套接字进行通讯有点浪费。
上图为单进程的socketpair
上图为父子进程分离后的socketpair
为了解决这个问题,Unix系统提供了无名套接字socketpair,不须要端口也能够建立套接字,父子进程经过socketpair来进行全双工通讯。
socketpair返回两个套接字对象,一个用于读一个用于写,它有点相似于pipe,只不过pipe返回的是两个文件描述符,都是整数。因此写起代码形式上跟pipe几乎没有什么区别。
咱们使用sock.send()和sock.recv()来对套接字进行读写,经过sock.close()来关闭套接字对象。
# coding: utf-8 import os import sys import math import socket def slice(mink, maxk): s = 0.0 for k in range(mink, maxk): s += 1.0/(2*k+1)/(2*k+1) return s def pi(n): childs = {} unit = n / 10 for i in range(10): # 分10个子进程 mink = unit * i maxk = mink + unit rsock, wsock = socket.socketpair() pid = os.fork() if pid > 0: childs[pid] = rsock wsock.close() else: rsock.close() s = slice(mink, maxk) # 子进程开始计算 wsock.send(str(s)) wsock.close() sys.exit(0) # 子进程结束 sums = [] for pid, rsock in childs.items(): sums.append(float(rsock.recv(1024))) rsock.close() os.waitpid(pid, 0) # 等待子进程结束 return math.sqrt(sum(sums) * 8) print pi(10000000)
输出
3.14159262176
相对于管道只能用于父子进程之间通讯,Unix还提供了有名管道可让任意进程进行通讯。有名管道又称fifo,它会将本身注册到文件系统里一个文件,参数通讯的进程经过读写这个文件进行通讯。 fifo要求读写双方必须同时打开才能够继续进行读写操做,不然打开操做会堵塞直到对方也打开。
# coding: utf-8 import os import sys import math def slice(mink, maxk): s = 0.0 for k in range(mink, maxk): s += 1.0/(2*k+1)/(2*k+1) return s def pi(n): childs = [] unit = n / 10 fifo_path = "/tmp/fifo_pi" os.mkfifo(fifo_path) # 建立named pipe for i in range(10): # 分10个子进程 mink = unit * i maxk = mink + unit pid = os.fork() if pid > 0: childs.append(pid) else: s = slice(mink, maxk) # 子进程开始计算 with open(fifo_path, "w") as ff: ff.write(str(s) + "\n") sys.exit(0) # 子进程结束 sums = [] while True: with open(fifo_path, "r") as ff: # 子进程关闭写端,读进程会收到eof # 因此必须循环打开,屡次读取 # 读够数量了就能够结束循环了 sums.extend([float(x) for x in ff.read(1024).strip().split("\n")]) if len(sums) == len(childs): break for pid in childs: os.waitpid(pid, 0) # 等待子进程结束 os.unlink(fifo_path) # 移除named pipe return math.sqrt(sum(sums) * 8) print pi(10000000)
输出
3.14159262176
操做系统也提供了跨进程的消息队列对象可让咱们直接使用,只不过python没有默认提供包装好的api来直接使用。咱们必须使用第三方扩展来完成OS消息队列通讯。第三方扩展是经过使用Python包装的C实现来完成的。
OS消息队列有两种形式,一种是posix消息队列,另外一种是systemv消息队列,有些操做系统二者都支持,有些只支持其中的一个,好比macos仅支持systemv消息队列,我本地的python的docker镜像是debian linux,它仅支持posix消息队列。
posix消息队列 咱们先使用posix消息队列来完成圆周率的计算,posix消息队列须要提供一个惟一的名称,它必须是/
开头。close()方法仅仅是减小内核消息队列对象的引用,而不是完全关闭它。unlink()方法才能完全销毁它。O_CREAT选项表示若是不存在就建立。向队列里塞消息使用send方法,收取消息使用receive方法,receive方法返回一个tuple,tuple的第一个值是消息的内容,第二个值是消息的优先级。之因此有优先级,是由于posix消息队列支持消息的排序,在send方法的第二个参数能够提供优先级整数值,默认为0,越大优先级越高。
# coding: utf-8 import os import sys import math from posix_ipc import MessageQueue as Queue def slice(mink, maxk): s = 0.0 for k in range(mink, maxk): s += 1.0/(2*k+1)/(2*k+1) return s def pi(n): pids = [] unit = n / 10 q = Queue("/pi", flags=os.O_CREAT) for i in range(10): # 分10个子进程 mink = unit * i maxk = mink + unit pid = os.fork() if pid > 0: pids.append(pid) else: s = slice(mink, maxk) # 子进程开始计算 q.send(str(s)) q.close() sys.exit(0) # 子进程结束 sums = [] for pid in pids: sums.append(float(q.receive()[0])) os.waitpid(pid, 0) # 等待子进程结束 q.close() q.unlink() # 完全销毁队列 return math.sqrt(sum(sums) * 8) print pi(10000000)
输出
3.14159262176
systemv消息队列 systemv消息队列和posix消息队列用起来有所不一样。systemv的消息队列是以整数key做为名称,若是不指定,它就建立一个惟一的未占用的整数key。它还提供消息类型的整数参数,可是不支持消息优先级。
# coding: utf-8 import os import sys import math import sysv_ipc from sysv_ipc import MessageQueue as Queue def slice(mink, maxk): s = 0.0 for k in range(mink, maxk): s += 1.0/(2*k+1)/(2*k+1) return s def pi(n): pids = [] unit = n / 10 q = Queue(key=None, flags=sysv_ipc.IPC_CREX) for i in range(10): # 分10个子进程 mink = unit * i maxk = mink + unit pid = os.fork() if pid > 0: pids.append(pid) else: s = slice(mink, maxk) # 子进程开始计算 q.send(str(s)) sys.exit(0) # 子进程结束 sums = [] for pid in pids: sums.append(float(q.receive()[0])) os.waitpid(pid, 0) # 等待子进程结束 q.remove() # 销毁消息队列 return math.sqrt(sum(sums) * 8) print pi(10000000)
输出
3.14159262176
共享内存也是很是常见的多进程通讯方式,操做系统负责将同一份物理地址的内存映射到多个进程的不一样的虚拟地址空间中。进而每一个进程均可以操做这分内存。考虑到物理内存的惟一性,它属于临界区资源,须要在进程访问时搞好并发控制,好比使用信号量。咱们经过一个信号量来控制全部子进程的顺序读写共享内存。咱们分配一个8字节double类型的共享内存用来存储极限的和,每次从共享内存中读出来时,要使用struct进行反序列化(unpack),将新的值写进去以前也要使用struct进行序列化(pack)。每次读写操做都须要将读写指针移动到内存开头位置(lseek)。
# coding: utf-8 import os import sys import math import struct import posix_ipc from posix_ipc import Semaphore from posix_ipc import SharedMemory as Memory def slice(mink, maxk): s = 0.0 for k in range(mink, maxk): s += 1.0/(2*k+1)/(2*k+1) return s def pi(n): pids = [] unit = n / 10 sem_lock = Semaphore("/pi_sem_lock", flags=posix_ipc.O_CREX, initial_value=1) # 使用一个信号量控制多个进程互斥访问共享内存 memory = Memory("/pi_rw", size=8, flags=posix_ipc.O_CREX) os.lseek(memory.fd, 0, os.SEEK_SET) # 初始化和为0.0的double值 os.write(memory.fd, struct.pack('d', 0.0)) for i in range(10): # 分10个子进程 mink = unit * i maxk = mink + unit pid = os.fork() if pid > 0: pids.append(pid) else: s = slice(mink, maxk) # 子进程开始计算 sem_lock.acquire() try: os.lseek(memory.fd, 0, os.SEEK_SET) bs = os.read(memory.fd, 8) # 从共享内存读出来当前值 cur_val, = struct.unpack('d', bs) # 反序列化,逗号不能少 cur_val += s # 加上当前进程的计算结果 bs = struct.pack('d', cur_val) # 序列化 os.lseek(memory.fd, 0, os.SEEK_SET) os.write(memory.fd, bs) # 写进共享内存 memory.close_fd() finally: sem_lock.release() sys.exit(0) # 子进程结束 sums = [] for pid in pids: os.waitpid(pid, 0) # 等待子进程结束 os.lseek(memory.fd, 0, os.SEEK_SET) bs = os.read(memory.fd, 8) # 读出最终这结果 sums, = struct.unpack('d', bs) # 反序列化 memory.close_fd() # 关闭共享内存 memory.unlink() # 销毁共享内存 sem_lock.unlink() # 销毁信号量 return math.sqrt(sums * 8) print pi(10000000)
输出
3.14159262176
阅读更多Python高级文章,请关注公众号「码洞」