多进程相关
Posted wangpanger
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了多进程相关相关的知识，希望对你有一定的参考价值。
#一 操作系统的作用： 1：隐藏丑陋复杂的硬件接口，提供良好的抽象接口 2：管理、调度进程，并且将多个进程对硬件的竞争变得有序 #二 多道技术： 1.产生背景：针对单核，实现并发 ps： 现在的主机一般是多核，那么每个核都会利用多道技术 有4个cpu，运行于cpu1的某个程序遇到io阻塞，会等到io结束再重新调度，会被调度到4个 cpu中的任意一个，具体由操作系统调度算法决定。 2.空间上的复用：如内存中同时有多道程序 3.时间上的复用：复用一个cpu的时间片 强调：遇到io切，占用cpu时间过长也切，核心在于切之前将进程的状态保存下来，这样 才能保证下次切换回来时，能基于上次切走的位置继续运行
一 什么是进程
进程（Process）是计算机中的程序关于某数据集合上的一次运行活动，是系统进行资源分配和调度的基本单位，是操作系统结构的基础。在早期面向进程设计的计算机结构中，进程是程序的基本执行实体；在当代面向线程设计的计算机结构中，进程是线程的容器。程序是指令、数据及其组织形式的描述，进程是程序的实体。我们自己在python文件中写了一些代码，这叫做程序，运行这个python文件的时候，这叫做进程。
狭义定义：进程是正在运行的程序的实例（an instance of a computer program that is being executed）。
广义定义：进程是一个具有一定独立功能的程序关于某个数据集合的一次运行活动。它是操作系统动态执行的基本单元，在传统的操作系统中，进程既是基本的分配单元，也是基本的执行单元。
第一，进程是一个实体。每一个进程都有它自己的地址空间，一般情况下，包括文本区域（text region）（python的文件）、数据区域（data region）（python文件中定义的一些变量数据）和堆栈（stack region）。文本区域存储处理器执行的代码；数据区域存储变量和进程执行期间使用的动态分配的内存；堆栈区域存储着活动过程调用的指令和本地变量。 第二，进程是一个“执行中的程序”。程序是一个没有生命的实体，只有处理器赋予程序生命时（操作系统执行之），它才能成为一个活动的实体，我们称其为进程。[3] 进程是操作系统中最基本、重要的概念。是多道程序系统出现后，为了刻画系统内部出现的动态情况，描述系统内部各道程序的活动规律引进的一个概念,所有多道程序设计操作系统都建立在进程的基础上。
动态性：进程的实质是程序在多道程序系统中的一次执行过程，进程是动态产生，动态消亡的。
并发性：任何进程都可以同其他进程一起并发执行
独立性：进程是一个能独立运行的基本单位，同时也是系统分配资源和调度的独立单位；
异步性：由于进程间的相互制约，使进程具有执行的间断性，即进程按各自独立的、不可预知的速度向前推进
结构特征：进程由程序、数据和进程控制块三部分组成。
多个不同的进程可以包含相同的程序：一个程序在不同的数据集里就构成不同的进程，能得到不同的结果；但是执行过程中，程序不能发生改变。
程序是指令和数据的有序集合，其本身没有任何运行的含义，是一个静态的概念。
而进程是程序在处理机上的一次执行过程，它是一个动态的概念。
程序可以作为一种软件资料长期存在，而进程是有一定生命期的。
程序是永久的，进程是暂时的。
举例：就像qq一样，qq是我们安装在自己电脑上的客户端程序，其实就是一堆的代码文件，我们不运行qq，那么他就是一堆代码程序，当我们运行qq的时候，这些代码运行起来，就成为一个进程了。
二 进程的调度
要想多个进程交替运行，操作系统必须对这些进程进行调度，这个调度也不是随即进行的，而是需要遵循一定的法则，由此就有了进程的调度算法。
先来先服务（FCFS）调度算法是一种最简单的调度算法，该算法既可用于作业调度，也可用于进程调度。FCFS算法比较有利于长作业（进程），而不利于短作业（进程）。由此可知，本算法适合于CPU繁忙型作业，而不利于I/O繁忙型的作业（进程）。
短作业（进程）优先调度算法（SJ/PF）是指对短作业或短进程优先调度的算法，该算法既可用于作业调度，也可用于进程调度。但其对长作业不利；不能保证紧迫性作业（进程）被及时处理；作业的长短只是被估算出来的。
时间片轮转(Round Robin，RR)法的基本思路是让每个进程在就绪队列中的等待时间与享受服务的时间成比例。在时间片轮转法中，需要将CPU的处理时间分成固定大小的时间片，例如，几十毫秒至几百毫秒。如果一个进程在被调度选中之后用完了系统规定的时间片，但又未完成要求的任务，则它自行释放自己所占有的CPU而排到就绪队列的末尾，等待下一次调度。同时，进程调度程序又去调度当前就绪队列中的第一个进程。 显然，轮转法只能用来调度分配一些可以抢占的资源。这些可以抢占的资源可以随时被剥夺，而且可以将它们再分配给别的进程。CPU是可抢占资源的一种。但打印机等资源是不可抢占的。由于作业调度是对除了CPU之外的所有系统硬件资源的分配，其中包含有不可抢占资源，所以作业调度不使用轮转法。 在轮转法中，时间片长度的选取非常重要。首先，时间片长度的选择会直接影响到系统的开销和响应时间。如果时间片长度过短，则调度程序抢占处理机的次数增多。这将使进程上下文切换次数也大大增加，从而加重系统开销。反过来，如果时间片长度选择过长，例如，一个时间片能保证就绪队列中所需执行时间最长的进程能执行完毕，则轮转法变成了先来先服务法。时间片长度的选择是根据系统对响应时间的要求和就绪队列中所允许最大的进程数来确定的。 在轮转法中，加入到就绪队列的进程有3种情况： 一种是分给它的时间片用完，但进程还未完成，回到就绪队列的末尾等待下次调度去继续执行。 另一种情况是分给该进程的时间片并未用完，只是因为请求I/O或由于进程的互斥与同步关系而被阻塞。当阻塞解除之后再回到就绪队列。 第三种情况就是新创建进程进入就绪队列。 如果对这些进程区别对待，给予不同的优先级和时间片从直观上看，可以进一步改善系统服务质量和效率。例如，我们可把就绪队列按照进程到达就绪队列的类型和进程被阻塞时的阻塞原因分成不同的就绪队列，每个队列按FCFS原则排列，各队列之间的进程享有不同的优先级，但同一队列内优先级相同。这样，当一个进程在执行完它的时间片之后，或从睡眠中被唤醒以及被创建之后，将进入不同的就绪队列。
前面介绍的各种用作进程调度的算法都有一定的局限性。如短进程优先的调度算法，仅照顾了短进程而忽略了长进程，而且如果并未指明进程的长度，则短进程优先和基于进程长度的抢占式调度算法都将无法使用。 而多级反馈队列调度算法则不必事先知道各种进程所需的执行时间，而且还可以满足各种类型进程的需要，因而它是目前被公认的一种较好的进程调度算法。在采用多级反馈队列调度算法的系统中，调度算法的实施过程如下所述。 (1) 应设置多个就绪队列，并为各个队列赋予不同的优先级。第一个队列的优先级最高，第二个队列次之，其余各队列的优先权逐个降低。该算法赋予各个队列中进程执行时间片的大小也各不相同，在优先权愈高的队列中，为每个进程所规定的执行时间片就愈小。例如，第二个队列的时间片要比第一个队列的时间片长一倍，……，第i+1个队列的时间片要比第i个队列的时间片长一倍。 (2) 当一个新进程进入内存后，首先将它放入第一队列的末尾，按FCFS原则排队等待调度。当轮到该进程执行时，如它能在该时间片内完成，便可准备撤离系统；如果它在一个时间片结束时尚未完成，调度程序便将该进程转入第二队列的末尾，再同样地按FCFS原则等待调度执行；如果它在第二队列中运行一个时间片后仍未完成，再依次将它放入第三队列，……，如此下去，当一个长作业(进程)从第一队列依次降到第n队列后，在第n 队列便采取按时间片轮转的方式运行。 (3) 仅当第一队列空闲时，调度程序才调度第二队列中的进程运行；仅当第1～(i-1)队列均空时，才会调度第i队列中的进程运行。如果处理机正在第i队列中为某进程服务时，又有新进程进入优先权较高的队列(第1～(i-1)中的任何一个队列)，则此时新进程将抢占正在运行进程的处理机，即由调度程序把正在运行的进程放回到第i队列的末尾，把处理机分配给新到的高优先权进程。
三 并行与并发
通过进程之间的调度，也就是进程之间的切换，我们用户感知到的好像是两个视频文件同时在播放，或者音乐和游戏同时在进行，那就让我们来看一下什么叫做并发和并行
无论是并行还是并发，在用户看来都是‘同时‘运行的，不管是进程还是线程，都只是一个任务而已，真是干活的是cpu，cpu来做这些任务，而一个cpu同一时刻只能执行一个任务
并发：是伪并行，即看起来是同时运行。单个cpu+多道技术就可以实现并发，（并行也属于并发）
并行：并行：同时运行，只有具备多个cpu才能实现并行
单核下，可以利用多道技术，多个核，每个核也都可以利用多道技术（多道技术是针对单核而言的）
有四个核，六个任务，这样同一时间有四个任务被执行，假设分别被分配给了cpu1，cpu2，cpu3，cpu4，
一旦任务1遇到I/O就被迫中断执行，此时任务5就拿到cpu1的时间片去执行，这就是单核下的多道技术
而一旦任务1的I/O结束了，操作系统会重新调用它(需知进程的调度、分配给哪个cpu运行，由操作系统说了算)，可能被分配给四个cpu中的任意一个去执行
多道技术概念回顾：内存中同时存入多道（多个）程序，cpu从一个进程快速切换到另外一个，使每个进程各自运行几十或几百毫秒，这样，虽然在某一个瞬间，一个cpu只能执行一个任务，但在1秒内，cpu却可以运行多个进程，这就给人产生了并行的错觉，即伪并行，以此来区分多处理器操作系统的真正硬件并行（多个cpu共享同一个物理内存）
四 同步异步阻塞非阻塞（重点）
1.进程状态介绍
在了解其他概念之前，我们首先要了解进程的几个状态。在程序运行的过程中，由于被操作系统的调度算法控制，程序会进入几个状态：就绪，运行和阻塞。
（1）就绪(Ready)状态
当进程已分配到除CPU以外的所有必要的资源，只要获得处理机便可立即执行，这时的进程状态称为就绪状态。
（2）执行/运行（Running）状态当进程已获得处理机，其程序正在处理机上执行，此时的进程状态称为执行状态。
（3）阻塞(Blocked)状态正在执行的进程，由于等待某个事件发生而无法执行时，便放弃处理机而处于阻塞状态。引起进程阻塞的事件可有多种，例如，等待I/O完成、申请缓冲区不能满足、等待信件(信号)等。
事件请求：input、sleep、文件输入输出、recv、accept等
事件发生：sleep、input等完成了
时间片到了之后有回到就绪状态，这三个状态不断的在转换。
2.同步异步
所谓同步就是一个任务的完成需要依赖另外一个任务时，只有等待被依赖的任务完成后，依赖的任务才能算完成，这是一种可靠的任务序列。要么成功都成功，失败都失败，两个任务的状态可以保持一致。其实就是一个程序结束才执行另外一个程序，串行的，不一定两个程序就有依赖关系。
所谓异步是不需要等待被依赖的任务完成，只是通知被依赖的任务要完成什么工作，依赖的任务也立即执行，只要自己完成了整个任务就算完成了。至于被依赖的任务最终是否真正完成，依赖它的任务无法确定，所以它是不可靠的任务序列。
3.阻塞与非阻塞
阻塞和非阻塞这两个概念与程序（线程）等待消息通知(无所谓同步或者异步)时的状态有关。也就是说阻塞与非阻塞主要是程序（线程）等待消息通知时的状态角度来说的
继续上面的那个例子，不论是排队还是使用号码等待通知，如果在这个等待的过程中，等待者除了等待消息通知之外不能做其它的事情，
那么该机制就是阻塞的，表现在程序中,也就是该程序一直阻塞在该函数调用处不能继续往下执行。相反，有的人喜欢在等待取餐的时候
一边打游戏一边等待，这样的状态就是非阻塞的，因为他(等待者)没有阻塞在这个消息通知上，而是一边做自己的事情一边等待。阻塞
的方法：input、time.sleep，socket中的recv、accept等等。
4.同步/异步 与 阻塞和非阻塞
- 同步阻塞形式
效率最低。拿上面的例子来说，就是你专心排队，什么别的事都不做。
- 异步阻塞形式
如果在排队取餐的人采用的是异步的方式去等待消息被触发（通知），也就是领了一张小纸条，假如在这段时间里他不能做其它的事情，就在那坐着等着，不能玩游戏等，那么很显然，这个人被阻塞在了这个等待的操作上面；
异步操作是可以被阻塞住的，只不过它不是在处理消息时阻塞，而是在等待消息通知时被阻塞。
- 同步非阻塞形式
实际上是效率低下的。
想象一下你一边打着电话一边还需要抬头看到底队伍排到你了没有，如果把打电话和观察排队的位置看成是程序的两个操作的话，这个程序需要在这两种不同的行为之间来回的切换，效率可想而知是低下的。
- 异步非阻塞形式
效率更高，
因为打电话是你(等待者)的事情，而通知你则是柜台(消息触发机制)的事情，程序没有在两种不同的操作中来回切换。
比如说，这个人突然发觉自己烟瘾犯了，需要出去抽根烟，于是他告诉点餐员说，排到我这个号码的时候麻烦到外面通知我一下，那么他就没有被阻塞在这个等待的操作上面，自然这个就是异步+非阻塞的方式了。
很多人会把同步和阻塞混淆，是因为很多时候同步操作会以阻塞的形式表现出来，同样的，很多人也会把异步和非阻塞混淆，因为异步操作一般都不会在真正的IO操作处被阻塞。
=======================================================
上面的内容都是进程的一些理论基础，下面的内容是python中进程的应用实战
===============================================================
五 multiprocess模块
仔细说来，multiprocess不是一个模块而是python中一个操作、管理进程的包。 之所以叫multi是取自multiple的多功能的意思,在这个包中几乎包含了和进程有关的所有子模块。由于提供的子模块非常多，为了方便大家归类记忆，我将这部分大致分为四个部分：创建进程部分，进程同步部分，进程池部分，进程之间数据共享。重点强调：进程没有任何共享状态，进程修改的数据，改动仅限于该进程内，但是通过一些特殊的方法，可以实现进程之间数据的共享。
1.process模块介绍
process模块是一个创建进程的模块，借助这个模块，就可以完成进程的创建。
Process([group [, target [, name [, args [, kwargs]]]]])，由该类实例化得到的对象，表示一个子进程中的任务（尚未启动）
强调：
1. 需要使用关键字的方式来指定参数
2. args指定的为传给target函数的位置参数，是一个元组形式，必须有逗号
我们先写一个程序来看看：
上面说了，我们通过主进程创建的子进程是异步执行的，那么我们就验证一下，并且看一下子进程和主进程(也就是父进程)的ID号（讲一下pid和ppid，使用pycharm举例），来看看是否是父子关系。
import time import os #os.getpid() 获取自己进程的ID号 #os.getppid() 获取自己进程的父进程的ID号 from multiprocessing import Process def func(): print(‘aaaa‘) time.sleep(1) print(‘子进程>>‘,os.getpid()) print(‘该子进程的父进程>>‘,os.getppid()) print(12345) if __name__ == ‘__main__‘: #首先我运行当前这个文件，运行的这个文件的程序，那么就产生了主进程 p = Process(target=func,) p.start() print(‘*‘ * 10) print(‘父进程>>‘,os.getpid()) print(‘父进程的父进程>>‘,os.getppid()) #加上time和进程号给大家看一看结果： #********** 首先打印出来了出进程的程序，然后打印的是子进程的，也就是子进程是异步执行的，相当于主进程和子进程同时运行着，如果是同步的话，我们先执行的是func()，然后再打印主进程最后的10个*号。 #父进程>> 3308 #父进程的父进程>> 5916 #我运行的test.py文件的父进程号，它是pycharm的进程号，看下面的截图 #aaaa #子进程>> 4536 #该子进程的父进程>> 3308 #是我主进程的ID号，说明主进程为它的父进程 #12345
打开windows下的任务管理器，看pycharm的pid进程号，是我们上面运行的test.py这个文件主进程的父进程号：
看一个问题，说明linux和windows两个不同的操作系统创建进程的不同机制导致的不同结果：
import time import os from multiprocessing import Process def func(): print(‘aaaa‘) time.sleep(1) print(‘子进程>>‘,os.getpid()) print(‘该子进程的父进程>>‘,os.getppid()) print(12345) print(‘太白老司机~~~~‘) #如果我在这里加了一个打印，你会发现运行结果中会出现两次打印出来的太白老司机，因为我们在主进程中开了一个子进程，子进程中的程序相当于import的主进程中的程序，那么import的时候会不会执行你import的那个文件的程序啊，前面学的，是会执行的，所以出现了两次打印 #其实是因为windows开起进程的机制决定的，在linux下是不存在这个效果的，因为windows使用的是process方法来开启进程，他就会拿到主进程中的所有程序，而linux下只是去执行我子进程中注册的那个函数，不会执行别的程序，这也是为什么在windows下要加上执行程序的时候， 要加上if __name__ == ‘__main__‘:，否则会出现子进程中运行的时候还开启子进程，那就出现无限循环的创建进程了，就报错了
一个进程的生命周期：如果子进程的运行时间长，那么等到子进程执行结束程序才结束，如果主进程的执行时间长，那么主进程执行结束程序才结束，实际上我们在子进程中打印的内容是在主进程的执行结果中看不出来的，但是pycharm帮我们做了优化，因为它会识别到你这是开的子进程，帮你把子进程中打印的内容打印到了显示台上。
如果说一个主进程运行完了之后，我们把pycharm关了，但是子进程还没有执行结束，那么子进程还存在吗？这要看你的进程是如何配置的，如果说我们没有配置说我主进程结束，子进程要跟着结束，那么主进程结束的时候，子进程是不会跟着结束的，他会自己执行完，如果我设定的是主进程结束，子进程必须跟着结束，那么就不会出现单独的子进程（孤儿进程）了，具体如何设置，看下面的守护进程的讲解。比如说，我们将来启动项目的时候，可能通过cmd来启动，那么我cmd关闭了你的项目就会关闭吗，不会的，因为你的项目不能停止对外的服务，对吧。
Process类中参数的介绍：
参数介绍：
1 group参数未使用，值始终为None
2 target表示调用对象，即子进程要执行的任务
3 args表示调用对象的位置参数元组，args=(1,2,‘egon‘,)
4 kwargs表示调用对象的字典,kwargs={‘name‘:‘egon‘,‘age‘:18}
5 name为子进程的名称
给要执行的函数传参数：
def func(x,y): print(x) time.sleep(1) print(y) if __name__ == ‘__main__‘: p = Process(target=func,args=(‘姑娘‘,‘来玩啊！‘))#这是func需要接收的参数的传送方式。 p.start() print(‘父进程执行结束！‘) #执行结果： 父进程执行结束！ 姑娘 来玩啊！
Process类中各方法的介绍：
1 p.start()：启动进程，并调用该子进程中的p.run()
2 p.run():进程启动时运行的方法，正是它去调用target指定的函数，我们自定义类的类中一定要实现该方法
3 p.terminate():强制终止进程p，不会进行任何清理操作，如果p创建了子进程，该子进程就成了僵尸进程，使用该方法需要特别小心这种情况。如果p还保存了一个锁那么也将不会被释放，进而导致死锁
4 p.is_alive():如果p仍然运行，返回True
5 p.join([timeout]):主线程等待p终止（强调：是主线程处于等的状态，而p是处于运行的状态）。timeout是可选的超时时间，需要强调的是，p.join只能join住start开启的进程，而不能join住run开启的进程
join方法的例子：
让主进程加上join的地方等待（也就是阻塞住），等待子进程执行完之后，再继续往下执行我的主进程，好多时候，我们主进程需要子进程的执行结果，所以必须要等待。join感觉就像是将子进程和主进程拼接起来一样，将异步改为同步执行。
def func(x,y): print(x) time.sleep(1) print(y) if __name__ == ‘__main__‘: p = Process(target=func,args=(‘姑娘‘,‘来玩啊！‘)) p.start() print(‘我这里是异步的啊！‘) #这里相对于子进程还是异步的 p.join() #只有在join的地方才会阻塞住，将子进程和主进程之间的异步改为同步 print(‘父进程执行结束！‘) #打印结果： 我这里是异步的啊！ 姑娘 来玩啊！ 父进程执行结束！
用for循环创建多个进程，并且要等所有子进程结束后再执行主进程时：
#下面的注释按照编号去看，别忘啦！ import time import os from multiprocessing import Process def func(x,y): print(x) # time.sleep(1) #进程切换：如果没有这个时间间隔，那么你会发现func执行结果是打印一个x然后一个y，再打印一个x一个y，不会出现打印多个x然后打印y的情况，因为两个打印距离太近了而且执行的也非常快，但是如果你这段程序运行慢的话，你就会发现进程之间的切换了。 print(y) if __name__ == ‘__main__‘: p_list= [] for i in range(10): p = Process(target=func,args=(‘姑娘%s‘%i,‘来玩啊！‘)) p_list.append(p) p.start() [ap.join() for ap in p_list] #4、这是解决办法，前提是我们的子进程全部都已经去执行了，那么我在一次给所有正在执行的子进程加上join，那么主进程就需要等着所有子进程执行结束才会继续执行自己的程序了，并且保障了所有子进程是异步执行的。 # p.join() #1、如果加到for循环里面，那么所有子进程包括父进程就全部变为同步了，因为for循环也是主进程的，循环第一次的时候，一个进程去执行了，然后这个进程就join住了，那么for循环就不会继续执行了，等着第一个子进程执行结束才会继续执行for循环去创建第二个子进程。 #2、如果我不想这样的，也就是我想所有的子进程是异步的，然后所有的子进程执行完了再执行主进程 #p.join() #3、如果这样写的话，多次运行之后，你会发现会出现主进程的程序比一些子进程先执行完，因为我们p.join()是对最后一个子进程进行了join，也就是说如果这最后一个子进程先于其他子进程执行完，那么主进程就会去执行，而此时如果还有一些子进程没有执行完，而主进程执行 #完了，那么就会先打印主进程的内容了，这个cpu调度进程的机制有关系，因为我们的电脑可能只有4个cpu，我的子进程加上住进程有11个，虽然我for循环是按顺序起进程的，但是操作系统一定会按照顺序给你执行你的进程吗，答案是不会的，操作系统会按照自己的算法来分配进 #程给cpu去执行，这里也解释了我们打印出来的子进程中的内容也是没有固定顺序的原因，因为打印结果也需要调用cpu，可以理解成进程在争抢cpu，如果同学你想问这是什么算法，这就要去研究操作系统啦。那我们的想所有子进程异步执行，然后再执行主进程的这个需求怎么解决啊 print(‘不要钱~~~~~~~~~~~~~~~~！‘)
模拟两个应用场景：1、同时对一个文件进行写操作 2、同时创建多个文件
import time import os import re from multiprocessing import Process #多进程同时对一个文件进行写操作 def func(x,y,i): with open(x,‘a‘,encoding=‘utf-8‘) as f: print(‘当前进程%s拿到的文件的光标位置>>%s‘%(os.getpid(),f.tell())) f.write(y) #多进程同时创建多个文件 # def func(x, y): # with open(x, ‘w‘, encoding=‘utf-8‘) as f: # f.write(y) if __name__ == ‘__main__‘: p_list= [] for i in range(10): p = Process(target=func,args=(‘can_do_girl_lists.txt‘,‘姑娘%s‘%i,i)) # p = Process(target=func,args=(‘can_do_girl_info%s.txt‘%i,‘姑娘电话0000%s‘%i)) p_list.append(p) p.start() [ap.join() for ap in p_list] #这就是个for循环，只不过用列表生成式的形式写的 with open(‘can_do_girl_lists.txt‘,‘r‘,encoding=‘utf-8‘) as f: data = f.read() all_num = re.findall(‘d+‘,data) #打开文件，统计一下里面有多少个数据，每个数据都有个数字，所以re匹配一下就行了 print(‘>>>>>‘,all_num,‘.....%s‘%(len(all_num))) #print([i in in os.walk(r‘你的文件夹路径‘)]) print(‘不要钱~~~~~~~~~~~~~~~~！‘)
Process类中自带封装的各属性的介绍
1 p.daemon：默认值为False，如果设为True，代表p为后台运行的守护进程，当p的父进程终止时，p也随之终止，并且设定为True后，p不能创建自己的新进程，必须在p.start()之前设置
2 p.name:进程的名称
3 p.pid：进程的pid
4 p.exitcode:进程在运行时为None、如果为–N，表示被信号N结束(了解即可)
5 p.authkey:进程的身份验证键,默认是由os.urandom()随机生成的32字符的字符串。这个键的用途是为涉及网络连接的底层进程间通信提供安全性，这类连接只有在具有相同的身份验证键时才能成功（了解即可）
2.Process类的使用
注意：在windows中Process()必须放到# if __name__ == ‘__main__‘:下
Since Windows has no fork, the multiprocessing module starts a new Python process and imports the calling module. If Process() gets called upon import, then this sets off an infinite succession of new processes (or until your machine runs out of resources). This is the reason for hiding calls to Process() inside if __name__ == "__main__" since statements inside this if-statement will not get called upon import. 由于Windows没有fork，多处理模块启动一个新的Python进程并导入调用模块。 如果在导入时调用Process（），那么这将启动无限继承的新进程（或直到机器耗尽资源）。 这是隐藏对Process（）内部调用的原，使用if __name__ == “__main __”，这个if语句中的语句将不会在导入时被调用。
进程的创建第二种方法（继承）
class MyProcess(Process): #自己写一个类，继承Process类 #我们通过init方法可以传参数，如果只写一个run方法，那么没法传参数，因为创建对象的是传参就是在init方法里面，面向对象的时候，我们是不是学过 def __init__(self,person): super().__init__() self.person=person def run(self): print(os.getpid()) print(self.pid) print(self.pid) print(‘%s 正在和女主播聊天‘ %self.person) # def start(self): # #如果你非要写一个start方法，可以这样写，并且在run方法前后，可以写一些其他的逻辑 # self.run() if __name__ == ‘__main__‘: p1=MyProcess(‘Jedan‘) p2=MyProcess(‘太白‘) p3=MyProcess(‘alexDSB‘) p1.start() #start内部会自动调用run方法 p2.start() # p2.run() p3.start() p1.join() p2.join() p3.join()
进程之间的数据是隔离的：
#我们说进程之间的数据是隔离的，也就是数据不共享，看下面的验证 from multiprocessing import Process n=100 #首先我定义了一个全局变量，在windows系统中应该把全局变量定义在if __name__ == ‘__main__‘之上就可以了 def work(): global n n=0 print(‘子进程内: ‘,n) if __name__ == ‘__main__‘: p=Process(target=work) p.start() p.join() #等待子进程执行完毕，如果数据共享的话，我子进程是不是通过global将n改为0了，但是你看打印结果，主进程在子进程执行结束之后，仍然是n=100，子进程n=0，说明子进程对n的修改没有在主进程中生效，说明什么？说明他们之间的数据是隔离的，互相不影响的 print(‘主进程内: ‘,n) #看结果： # 子进程内: 0 # 主进程内: 100
3.守护进程
之前我们讲的子进程是不会随着主进程的结束而结束，子进程全部执行完之后，程序才结束，那么如果有一天我们的需求是我的主进程结束了，由我主进程创建的那些子进程必须跟着结束，怎么办？守护进程就来了！
主进程创建守护进程
其一：守护进程会在主进程代码执行结束后就终止
其二：守护进程内无法再开启子进程,否则抛出异常：AssertionError: daemonic processes are not allowed to have children
注意：进程之间是互相独立的，主进程代码运行结束，守护进程随即终止
import os import time from multiprocessing import Process class Myprocess(Process): def __init__(self,person): super().__init__() self.person = person def run(self): print(os.getpid(),self.name) print(‘%s正在和女主播聊天‘ %self.person) time.sleep(3) if __name__ == ‘__main__‘: p=Myprocess(‘太白‘) p.daemon=True #一定要在p.start()前设置,设置p为守护进程,禁止p创建子进程,并且父进程代码执行结束,p即终止运行 p.start() # time.sleep(1) # 在sleep时linux下查看进程id对应的进程ps -ef|grep id print(‘主‘)
4.进程同步（锁）
通过刚刚的学习，我们千方百计实现了程序的异步，让多个任务可以同时在几个进程中并发处理，他们之间的运行没有顺序，一旦开启也不受我们控制。尽管并发编程让我们能更加充分的利用IO资源，但是也给我们带来了新的问题：进程之间数据不共享,但是共享同一套文件系统,所以访问同一个文件,或同一个打印终端,是没有问题的，而共享带来的是竞争，竞争带来的结果就是错乱，如何控制，就是加锁处理。
import os import time import random from multiprocessing import Process def work(n): print(‘%s: %s is running‘ %(n,os.getpid())) time.sleep(random.random()) print(‘%s:%s is done‘ %(n,os.getpid())) if __name__ == ‘__main__‘: for i in range(5): p=Process(target=work,args=(i,)) p.start() # 看结果：通过结果可以看出两个问题：问题一：每个进程中work函数的第一个打印就不是按照我们for循环的0-4的顺序来打印的 #问题二：我们发现，每个work进程中有两个打印，但是我们看到所有进程中第一个打印的顺序为0-2-1-4-3，但是第二个打印没有按照这个顺序，变成了2-1-0-3-4，说明我们一个进程中的程序的执行顺序都混乱了。 #问题的解决方法，第二个问题加锁来解决，第一个问题是没有办法解决的，因为进程开到了内核，有操作系统来决定进程的调度，我们自己控制不了 # 0: 9560 is running # 2: 13824 is running # 1: 7476 is running # 4: 11296 is running # 3: 14364 is running # 2:13824 is done # 1:7476 is done # 0:9560 is done # 3:14364 is done # 4:11296 is done
#由并发变成了串行,牺牲了运行效率,但避免了竞争 from multiprocessing import Process,Lock import os,time def work(n,lock): #加锁，保证每次只有一个进程在执行锁里面的程序，这一段程序对于所有写上这个锁的进程，大家都变成了串行 lock.acquire() print(‘%s: %s is running‘ %(n,os.getpid())) time.sleep(1) print(‘%s:%s is done‘ %(n,os.getpid())) #解锁，解锁之后其他进程才能去执行自己的程序 lock.release() if __name__ == ‘__main__‘: lock=Lock() for i in range(5): p=Process(target=work,args=(i,lock)) p.start() #打印结果： # 2: 10968 is running # 2:10968 is done # 0: 7932 is running # 0:7932 is done # 4: 4404 is running # 4:4404 is done # 1: 12852 is running # 1:12852 is done # 3: 980 is running # 3:980 is done #结果分析：（自己去多次运行一下，看看结果，我拿出其中一个结果来看）通过结果我们可以看出，多进程刚开始去执行的时候，每次运行，首先打印出来哪个进程的程序是不固定的，但是我们解决了上面打印混乱示例代码的第二个问题，那就是同一个进程中的两次打印都是先完成的，然后才切换到下一个进程去，打印下一个进程中的两个打印结果，说明我们控制住了同一进程中的代码执行顺序，如果涉及到多个进程去操作同一个数据或者文件的时候，就不担心数据算错或者文件中的内容写入混乱了。
上面这种情况虽然使用加锁的形式实现了顺序的执行，但是程序又重新变成串行了，这样确实会浪费了时间，却保证了数据的安全。
接下来，我们以模拟抢票为例，来看看数据安全的重要性。
#注意：首先在当前文件目录下创建一个名为db的文件 #文件db的内容为：{"count":1}，只有这一行数据,并且注意，每次运行完了之后，文件中的1变成了0，你需要手动将0改为1，然后在去运行代码。 #注意一定要用双引号，不然json无法识别 #并发运行，效率高，但竞争写同一文件，数据写入错乱 from multiprocessing import Process,Lock import time,json,random #查看剩余票数 def search(): dic=json.load(open(‘db‘)) #打开文件，直接load文件中的内容，拿到文件中的包含剩余票数的字典 print(‘ 33[43m剩余票数%s 33[0m‘ %dic[‘count‘]) #抢票 def get(): dic=json.load(open(‘db‘)) time.sleep(0.1) #模拟读数据的网络延迟，那么进程之间的切换，导致所有人拿到的字典都是{"count": 1}，也就是每个人都拿到了这一票。 if dic[‘count‘] >0: dic[‘count‘]-=1 time.sleep(0.2) #模拟写数据的网络延迟 json.dump(dic,open(‘db‘,‘w‘)) #最终结果导致，每个人显示都抢到了票，这就出现了问题~ print(‘ 33[43m购票成功 33[0m‘) def task(): search() get() if __name__ == ‘__main__‘: for i in range(3): #模拟并发100个客户端抢票 p=Process(target=task) p.start() #看结果分析：由于网络延迟等原因使得进程切换，导致每个人都抢到了这最后一张票 # 剩余票数1 # 剩余票数1 # 剩余票数1 # 购票成功 # 购票成功 # 购票成功
#注意：首先在当前文件目录下创建一个名为db的文件 #文件db的内容为：{"count":1}，只有这一行数据,并且注意，每次运行完了之后，文件中的1变成了0，你需要手动将0改为1，然后在去运行代码。 #注意一定要用双引号，不然json无法识别 #加锁保证数据安全，不出现混乱 from multiprocessing import Process,Lock import time,json,random #查看剩余票数 def search(): dic=json.load(open(‘db‘)) #打开文件，直接load文件中的内容，拿到文件中的包含剩余票数的字典 print(‘ 33[43m剩余票数%s 33[0m‘ %dic[‘count‘]) #抢票 def get(): dic=json.load(open(‘db‘)) time.sleep(0.1) #模拟读数据的网络延迟，那么进程之间的切换，导致所有人拿到的字典都是{"count": 1}，也就是每个人都拿到了这一票。 if dic[‘count‘] >0: dic[‘count‘]-=1 time.sleep(0.2) #模拟写数据的网络延迟 json.dump(dic,open(‘db‘,‘w‘)) #最终结果导致，每个人显示都抢到了票，这就出现了问题~ print(‘ 33[43m购票成功 33[0m‘) else: print(‘sorry，没票了亲！‘) def task(lock): search() #因为抢票的时候是发生数据变化的时候，所有我们将锁加加到这里 lock.acquire() get() lock.release() if __name__ == ‘__main__‘: lock = Lock() #创建一个锁 for i in range(3): #模拟并发100个客户端抢票 p=Process(target=task,args=(lock,)) #将锁作为参数传给task函数 p.start() #看结果分析：只有一个人抢到了票 # 剩余票数1 # 剩余票数1 # 剩余票数1 # 购票成功 #幸运的人儿 # sorry，没票了亲！ # sorry，没票了亲！
进程锁总结： 复制代码 #加锁可以保证多个进程修改同一块数据时，同一时间只能有一个任务可以进行修改，即串行的修改，没错，速度是慢了，但牺牲了速度却保证了数据安全。 虽然可以用文件共享数据实现进程间通信，但问题是： 1.效率低（共享数据基于文件，而文件是硬盘上的数据） 2.需要自己加锁处理 #因此我们最好找寻一种解决方案能够兼顾：1、效率高（多个进程共享一块内存的数据）2、帮我们处理好锁问题。这就是mutiprocessing模块为我们提供的基于消息的IPC通信机制：队列和管道。 队列和管道都是将数据存放于内存中 队列又是基于（管道+锁）实现的，可以让我们从复杂的锁问题中解脱出来， 我们应该尽量避免使用共享数据，尽可能使用消息传递和队列，避免处理复杂的同步和锁问题，而且在进程数目增多时，往往可以获得更好的可获展性。 IPC通信机制（了解）：IPC是intent-Process Communication的缩写，含义为进程间通信或者跨进程通信，是指两个进程之间进行数据交换的过程。IPC不是某个系统所独有的，任何一个操作系统都需要有相应的IPC机制， 比如Windows上可以通过剪贴板、管道和邮槽等来进行进程间通信，而Linux上可以通过命名共享内容、信号量等来进行进程间通信。android它也有自己的进程间通信方式，Android建构在Linux基础上，继承了一 部分Linux的通信方式。
5.队列（推荐使用）
进程彼此之间互相隔离，要实现进程间通信（IPC），multiprocessing模块支持两种形式：队列和管道，这两种方式都是使用消息传递的。队列就像一个特殊的列表，但是可以设置固定长度，并且从前面插入数据，从后面取出数据，先进先出。
Queue([maxsize]) 创建共享的进程队列。
参数 ：maxsize是队列中允许的最大项数。如果省略此参数，则无大小限制。
底层队列使用管道和锁实现。
先看下面的代码示例，然后再看方法介绍。
queue的方法介绍
q = Queue([maxsize])
创建共享的进程队列。maxsize是队列中允许的最大项数。如果省略此参数，则无大小限制。底层队列使用管道和锁定实现。另外，还需要运行支持线程以便队列中的数据传输到底层管道中。
Queue的实例q具有以下方法：
q.get( [ block [ ,timeout ] ] )
返回q中的一个项目。如果q为空，此方法将阻塞，直到队列中有项目可用为止。block用于控制阻塞行为，默认为True. 如果设置为False，将引发Queue.Empty异常（定义在Queue模块中）。timeout是可选超时时间，用在阻塞模式中。如果在制定的时间间隔内没有项目变为可用，将引发Queue.Empty异常。
q.get_nowait( )
同q.get(False)方法。
q.put(item [, block [,timeout ] ] )
将item放入队列。如果队列已满，此方法将阻塞至有空间可用为止。block控制阻塞行为，默认为True。如果设置为False，将引发Queue.Empty异常（定义在Queue库模块中）。timeout指定在阻塞模式中等待可用空间的时间长短。超时后将引发Queue.Full异常。
q.qsize()
返回队列中目前项目的正确数量。此函数的结果并不可靠，因为在返回结果和在稍后程序中使用结果之间，队列中可能添加或删除了项目。在某些系统上，此方法可能引发NotImplementedError异常。
q.empty()
如果调用此方法时 q为空，返回True。如果其他进程或线程正在往队列中添加项目，结果是不可靠的。也就是说，在返回和使用结果之间，队列中可能已经加入新的项目。
q.full()
如果q已满，返回为True. 由于线程的存在，结果也可能是不可靠的（参考q.empty（）方法）。。
代码示例：
from multiprocessing import Queue q=Queue(3) #创建一个队列对象，队列长度为3 #put ,get ,put_nowait,get_nowait,full,empty q.put(3) #往队列中添加数据 q.put(2) q.put(1) # q.put(4) # 如果队列已经满了，程序就会停在这里，等待数据被别人取走，再将数据放入队列。 # 如果队列中的数据一直不被取走，程序就会永远停在这里。 try: q.put_nowait(4) # 可以使用put_nowait，如果队列满了不会阻塞，但是会因为队列满了而报错。 except: # 因此我们可以用一个try语句来处理这个错误。这样程序不会一直阻塞下去，但是会丢掉这个消息。 print(‘队列已经满了‘) # 因此，我们再放入数据之前，可以先看一下队列的状态，如果已经满了，就不继续put了。 print(q.full()) #查看是否满了，满了返回True，不满返回False print(q.get()) #取出数据 print(q.get()) print(q.get()) # print(q.get()) # 同put方法一样，如果队列已经空了，那么继续取就会出现阻塞。 try: q.get_nowait(3) # 可以使用get_nowait，如果队列满了不会阻塞，但是会因为没取到值而报错。 except: # 因此我们可以用一个try语句来处理这个错误。这样程序不会一直阻塞下去。 print(‘队列已经空了‘) print(q.empty()) #空了
#看下面的队列的时候，按照编号看注释 import time from multiprocessing import Process, Queue # 8. q = Queue(2) #创建一个Queue对象，如果写在这里，那么在windows还子进程去执行的时候，我们知道子进程中还会执行这个代码，但是子进程中不能够再次创建了，也就是这个q就是你主进程中创建的那个q，通过我们下面在主进程中先添加了一个字符串之后，在去开启子进程，你会发现，小鬼这个字符串还在队列中，也就是说，我们使用的还是主进程中创建的这个队列。 def f(q): # q = Queue() #9. 我们在主进程中开启了一个q，如果我们在子进程中的函数里面再开一个q，那么你下面q.put(‘姑娘，多少钱~‘)添加到了新创建的这q里里面了 q.put(‘姑娘，多少钱~‘) #4.调用主函数中p进程传递过来的进程参数 put函数为向队列中添加一条数据。 # print(q.qsize()) #6.查看队列中有多少条数据了 def f2(q): print(‘》》》》》》》》‘) print(q.get()) #5.取数据 if __name__ == ‘__main__‘: q = Queue() #1.创建一个Queue对象 q.put(‘小鬼‘) p = Process(target=f, args=(q,)) #2.创建一个进程 p2 = Process(target=f2, args=(q,)) #3.创建一个进程 p.start() p2.start() time.sleep(1) #7.如果阻塞一点时间，就会出现主进程运行太快，导致我们在子进程中查看qsize为1个。 # print(q.get()) #结果：小鬼 print(q.get()) #结果：姑娘，多少钱~ p.join()
由生产者消费者模型我们发现一个问题：主进程永远不会结束，原因是：生产者p在生产完后就结束了，但是消费者c在取空了q之后，则一直处于死循环中且卡在q.get()这一步。解决方式无非是让生产者在生产完毕后，往队列中再发一个结束信号，这样消费者在接收到结束信号后就可以break出死循环。在有多个生产者和多个消费者时，由于队列我们说了是进程安全的，我一个进程拿走了结束信号，另外一个进程就拿不到了，还需要多发送一个结束信号，有几个取数据的进程就要发送几个结束信号，我们则需要用一个很low的方式去解决，其实我们的思路无非是发送结束信号而已，有另外一种队列提供了这种机制
JoinableQueue([maxsize])
#JoinableQueue([maxsize])：这就像是一个Queue对象，但队列允许项目的使用者通知生成者项目已经被成功处理。通知进程是使用共享的信号和条件变量来实现的。 #参数介绍： maxsize是队列中允许最大项数，省略则无大小限制。 #方法介绍： JoinableQueue的实例p除了与Queue对象相同的方法之外还具有： q.task_done()：使用者使用此方法发出信号，表示q.get()的返回项目已经被处理。如果调用此方法的次数大于从队列中删除项目的数量，将引发ValueError异常 q.join():生产者调用此方法进行阻塞，直到队列中所有的项目均被处理。阻塞将持续到队列中的每个项目均调用q.task_done（）方法为止，也就是队列中的数据全部被get拿走了。
from multiprocessing import Process,JoinableQueue import time,random,os def consumer(q): while True: res=q.get() # time.sleep(random.randint(1,3)) time.sleep(random.random()) print(‘ 33[45m%s 吃 %s 33[0m‘ %(os.getpid(),res)) q.task_done() #向q.join()发送一次信号,证明一个数据已经被取走并执行完了 def producer(name,q): for i in range(10): # time.sleep(random.randint(1,3)) time.sleep(random.random()) res=‘%s%s‘ %(name,i) q.put(res) print(‘ 33[44m%s 生产了 %s 33[0m‘ %(os.getpid(),res)) print(‘%s生产结束‘%name) q.join() #生产完毕，使用此方法进行阻塞，直到队列中所有项目均被处理。 print(‘%s生产结束~~~~~~‘%name) if __name__ == ‘__main__‘: q=JoinableQueue() #生产者们:即厨师们 p1=Process(target=producer,args=(‘包子‘,q)) p2=Process(target=producer,args=(‘骨头‘,q)) p3=Process(target=producer,args=(‘泔水‘,q)) #消费者们:即吃货们 c1=Process(target=consumer,args=(q,)) c2=Process(target=consumer,args=(q,)) c1.daemon=True #如果不加守护，那么主进程结束不了，但是加了守护之后，必须确保生产者的内容生产完并且被处理完了，所有必须还要在主进程给生产者设置join，才能确保生产者生产的任务被执行完了，并且能够确保守护进程在所有任务执行完成之后才随着主进程的结束而结束。 c2.daemon=True #开始 p_l=[p1,p2,p3,c1,c2] for p in p_l: p.start() p1.join() #我要确保你的生产者进程结束了，生产者进程的结束标志着你生产的所有的人任务都已经被处理完了 p2.join() p3.join() print(‘主‘) # 主进程等--->p1,p2,p3等---->c1,c2 # p1,p2,p3结束了,证明c1,c2肯定全都收完了p1,p2,p3发到队列的数据 # 因而c1,c2也没有存在的价值了,不需要继续阻塞在进程中影响主进程了。应该随着主进程的结束而结束,所以设置成守护进程就可以了。
6.管道（了解）
进程间通信（IPC）方式二：管道（不推荐使用，了解即可），会导致数据不安全的情况出现，后面我们会说到为什么会带来数据 不安全的问题。
#创建管道的类： Pipe([duplex]):在进程之间创建一条管道，并返回元组（conn1,conn2）,其中conn1，conn2表示管道两端的连接对象，强调一点：必须在产生Process对象之前产生管道 #参数介绍： dumplex:默认管道是全双工的，如果将duplex射成False，conn1只能用于接收，conn2只能用于发送。 #主要方法： conn1.recv():接收conn2.send(obj)发送的对象。如果没有消息可接收，recv方法会一直阻塞。如果连接的另外一端已经关闭，那么recv方法会抛出EOFError。 conn1.send(obj):通过连接发送对象。obj是与序列化兼容的任意对象 #其他方法： conn1.close():关闭连接。如果conn1被垃圾回收，将自动调用此方法 conn1.fileno():返回连接使用的整数文件描述符 conn1.poll([timeout]):如果连接上的数据可用，返回True。timeout指定等待的最长时限。如果省略此参数，方法将立即返回结果。如果将timeout射成None，操作将无限期地等待数据到达。 conn1.recv_bytes([maxlength]):接收c.send_bytes()方法发送的一条完整的字节消息。maxlength指定要接收的最大字节数。如果进入的消息，超过了这个最大值，将引发IOError异常，并且在连接上无法进行进一步读取。如果连接的另外一端已经关闭，再也不存在任何数据，将引发EOFError异常。 conn.send_bytes(buffer [, offset [, size]])：通过连接发送字节数据缓冲区，buffer是支持缓冲区接口的任意对象，offset是缓冲区中的字节偏移量，而size是要发送字节数。结果数据以单条消息的形式发出，然后调用c.recv_bytes()函数进行接收 conn1.recv_bytes_into(buffer [, offset]):接收一条完整的字节消息，并把它保存在buffer对象中，该对象支持可写入的缓冲区接口（即bytearray对象或类似的对象）。offset指定缓冲区中放置消息处的字节位移。返回值是收到的字节数。如果消息长度大于可用的缓冲区空间，将引发BufferTooShort异常。
from multiprocessing import Process, Pipe def f(conn): conn.send("Hello 妹妹") #子进程发送了消息 conn.close() if __name__ == ‘__main__‘: parent_conn, child_conn = Pipe() #建立管道，拿到管道的两端，双工通信方式，两端都可以收发消息 p = Process(target=f, args=(child_conn,)) #将管道的一段给子进程 p.start() #开启子进程 print(parent_conn.recv()) #主进程接受了消息 p.join()
应该特别注意管道端点的正确管理问题。如果是生产者或消费者中都没有使用管道的某个端点，就应将它关闭。这也说明了为何在生产者中关闭了管道的输出端，在消费者中关闭管道的输入端。如果忘记执行这些步骤，程序可能在消费者中的recv（）操作上挂起（就是阻塞）。管道是由操作系统进行引用计数的，必须在所有进程中关闭管道的相同一端就会能生成EOFError异常。因此，在生产者中关闭管道不会有任何效果，除非消费者也关闭了相同的管道端点。
from multiprocessing import Process, Pipe def f(parent_conn,child_conn): #parent_conn.close() #不写close将不会引发EOFError while True: try: print(child_conn.recv()) except EOFError: child_conn.close() break if __name__ == ‘__main__‘: parent_conn, child_conn = Pipe() p = Process(target=f, args=(parent_conn,child_conn,)) p.start() child_conn.close() parent_conn.send(‘hello‘) parent_conn.close() p.join()
主进程将管道的两端都传送给子进程，子进程和主进程共用管道的两种报错情况，都是在recv接收的时候报错的：
1.主进程和子进程中的管道的相同一端都关闭了，出现EOFError；
2.如果你管道的一端在主进程和子进程中都关闭了，但是你还用这个关闭的一端去接收消息，那么就会出现OSError；
所以你关闭管道的时候，就容易出现问题，需要将所有只用这个管道的进程中的两端全部关闭才行。当然也可以通过异常捕获（try：except EOFerror）来处理。
虽然我们在主进程和子进程中都打印了一下conn1一端的对象，发现两个不再同一个地址，但是子进程中的管道和主进程中的管道还是可以通信的，因为管道是同一套，系统能够记录。
我们的目的就是关闭所有的管道，那么主进程和子进程进行通信的时候，可以给子进程传管道的一端就够了，并且用我们之前学到的，信息发送完之后，再发送一个结束信号None，那么你收到的消息为None的时候直接结束接收或者说结束循环，就不用每次都关闭各个进程中的管道了。
from multiprocessing import Pipe,Process def func(conn): while True: msg = conn.recv() if msg is None:break print(msg) if __name__ == ‘__main__‘: conn1,conn2 = Pipe() p = Process(target=func,args=(conn1,)) p.start() for i in range(10): conn2.send(‘约吧‘) conn2.send(None)
from multiprocessing import Process,Pipe def consumer(p,name): produce, consume=p produce.close() while True: try: baozi=consume.recv() print(‘%s 收到包子:%s‘ %(name,baozi)) except EOFError: break def producer(seq,p): produce, consume=p consume.close() for i in seq: produce.send(i) if __name__ == ‘__main__‘: produce,consume=Pipe() c1=Process(target=consumer,args=((produce,consume),‘c1‘)) c1.start() seq=(i for i in range(10)) producer(seq,(produce,consume)) produce.close() consume.close() c1.join() print(‘主进程‘)
关于管道会造成数据不安全问题的官方解释： The two connection objects returned by Pipe() represent the two ends of the pipe. Each connection object has send() and recv() methods (among others). Note that data in a pipe may become corrupted if two processes (or threads) try to read from or write to the same end of the pipe at the same time. Of course there is no risk of corruption from processes using different ends of the pipe at the same time. 由Pipe方法返回的两个连接对象表示管道的两端。每个连接对象都有send和recv方法（除其他之外）。注意，如果两个进程（或线程）试图同时从管道的同一端读取或写入数据，那么管道中的数据可能会损坏。当然，在使用管道的不同端部的过程中不存在损坏风险。
from multiprocessing import Process,Pipe,Lock def consumer(p,name,lock): produce, consume=p produce.close() while True: lock.acquire() baozi=consume.recv() lock.release() if baozi: print(‘%s 收到包子:%s‘ %(name,baozi)) else: consume.close() break def producer(p,n): produce, consume=p consume.close() for i in range(n): produce.send(i) produce.send(None) produce.send(None) produce.close() if __name__ == ‘__main__‘: produce,consume=Pipe() lock = Lock() c1=Process(target=consumer,args=((produce,consume),‘c1‘,lock)) c2=Process(target=consumer,args=((produce,consume),‘c2‘,lock)) p1=Process(target=producer,args=((produce,consume),10)) c1.start() c2.start() p1.start() produce.close() consume.close() c1.join() c2.join() p1.join() print(‘主进程‘)
管道可以用于双工通信，通常利用在客户端/服务端中使用的请求/响应模型，或者远程过程调用，就可以使用管道编写与进程交互的程序，像前面将网络通信的时候，我们使用了一个叫subprocess的模块，里面有个参数是pipe管道，执行系统指令，并通过管道获取结果。
7.数据共享（了解）
展望未来，基于消息传递的并发编程是大势所趋
即便是使用线程，推荐做法也是将程序设计为大量独立的线程集合
通过消息队列交换数据。这样极大地减少了对使用锁定和其他同步手段的需求，还可以扩展到分布式系统中
进程间应该尽量避免通信，即便需要通信，也应该选择进程安全的工具来避免加锁带来的问题，应该尽量避免使用本节所讲的共享数据的方式，以后我们会尝试使用数据库来解决进程之间的数据共享问题。
进程之间数据共享的模块之一Manager模块：
进程间数据是独立的，可以借助于队列或管道实现通信，二者都是基于消息传递的
虽然进程间数据独立，但可以通过Manager实现数据共享，事实上Manager的功能远不止于此
多进程共同去处理共享数据的时候，就和我们多进程同时去操作一个文件中的数据是一样的，不加锁就会出现错误的结果，进程不安全的，所以也需要加
from multiprocessing import Manager,Process,Lock def work(d,lock): with lock: #不加锁而操作共享的数据,肯定会出现数据错乱 d[‘count‘]-=1 if __name__ == ‘__main__‘: lock=Lock() with Manager() as m: dic=m.dict({‘count‘:100}) p_l=[] for i in range(100): p=Process(target=work,args=(dic,lock)) p_l.append(p) p.start() for p in p_l: p.join() print(dic)
8.信号量（了解）
互斥锁同时只允许一个线程更改数据，而信号量Semaphore是同时允许一定数量的线程更改数据 。
假设商场里有4个迷你唱吧，所以同时可以进去4个人，如果来了第五个人就要在外面等待，等到有人出来才能再进去玩。
实现：
信号量同步基于内部计数器，每调用一次acquire()，计数器减1；每调用一次release()，计数器加1.当计数器为0时，acquire()调用被阻塞。这是迪科斯彻（Dijkstra）信号量概念P()和V()的Python实现。信号量同步机制适用于访问像服务器这样的有限资源。
信号量与进程池的概念很像，但是要区分开，信号量涉及到加锁的概念
from multiprocessing import Process,Semaphore import time,random def go_ktv(sem,user): sem.acquire() print(‘%s 占到一间ktv小屋‘ %user) time.sleep(random.randint(0,3)) #模拟每个人在ktv中待的时间不同 sem.release() if __name__ == ‘__main__‘: sem=Semaphore(4) p_l=[] for i in range(13): p=Process(target=go_ktv,args=(sem,‘user%s‘ %i,)) p.start() p_l.append(p) for i in p_l: i.join() print(‘============》‘)
9.事件（了解）
python线程的事件用于主线程控制其他线程的执行，事件主要提供了三个方法 set、wait、clear。
事件处理的机制：全局定义了一个“Flag”，如果“Flag”值为 False，那么当程序执行 event.wait 方法时就会阻塞，如果“Flag”值为True，那么event.wait 方法时便不再阻塞。
clear：将“Flag”设置为False
set：将“Flag”设置为True
from multiprocessing import Process,Semaphore,Event import time,random e = Event() #创建一个事件对象 print(e.is_set()) #is_set()查看一个事件的状态，默认为False，可通过set方法改为True print(‘look here！‘) # e.set() #将is_set()的状态改为True。 # print(e.is_set())#is_set()查看一个事件的状态，默认为False，可通过set方法改为Tr # e.clear() #将is_set()的状态改为False # print(e.is_set())#is_set()查看一个事件的状态，默认为False，可通过set方法改为Tr e.wait() #根据is_set()的状态结果来决定是否在这阻塞住，is_set()=False那么就阻塞，is_set()=True就不阻塞 print(‘give me！！‘) #set和clear 修改事件的状态 set-->True clear-->False #is_set 用来查看一个事件的状态 #wait 依据事件的状态来决定是否阻塞 False-->阻塞 True-->不阻塞
from multiprocessing import Process, Event import time, random def car(e, n): while True: if not e.is_set(): # 进程刚开启，is_set()的值是Flase，模拟信号灯为红色 print(‘ 33[31m红灯亮 33[0m，car%s等着‘ % n) e.wait() # 阻塞，等待is_set()的值变成True，模拟信号灯为绿色 print(‘ 33[32m车%s 看见绿灯亮了 33[0m‘ % n) time.sleep(random.randint(2,4)) if not e.is_set(): #如果is_set()的值是Flase，也就是红灯，仍然回到while语句开始 continue print(‘车开远了,car‘, n) break # def police_car(e, n): # while True: # if not e.is_set():# 进程刚开启，is_set()的值是Flase，模拟信号灯为红色 # print(‘ 33[31m红灯亮 33[0m，car%s等着‘ % n) # e.wait(0.1) # 阻塞，等待设置等待时间，等待0.1s之后没有等到绿灯就闯红灯走了 # if not e.is_set(): # print(‘ 33[33m红灯,警车先走 33[0m，car %s‘ % n) # else: # print(‘ 33[33;46m绿灯，警车走 33[0m，car %s‘ % n) # break def traffic_lights(e, inverval): while True: time.sleep(inverval) if e.is_set(): print(‘######‘, e.is_set()) e.clear() # ---->将is_set()的值设置为False else: e.set() # ---->将is_set()的值设置为True print(‘***********‘,e.is_set()) if __name__ == ‘__main__‘: e = Event() for i in range(10): p=Process(target=car,args=(e,i,)) # 创建10个进程控制10辆车 time.sleep(random.random(1, 3)) #车不是一下子全过来 p.start() # for i in range(5): # p = Process(target=police_car, args=(e, i,)) # 创建5个进程控制5辆警车 # p.start() #信号灯必须是单独的进程，因为它不管你车开到哪了，我就按照我红绿灯的规律来闪烁变换，对吧 t = Process(target=traffic_lights, args=(e, 5)) # 创建一个进程控制红绿灯 t.start() print(‘预备~~~~开始！！！‘)
六 进程池和mutiprocess.Poll
为什么要有进程池?进程池的概念。
在程序实际处理问题过程中，忙时会有成千上万的任务需要被执行，闲时可能只有零星任务。那么在成千上万个任务需要被执行的时候，我们就需要去创建成千上万个进程么？首先，创建进程需要消耗时间，销毁进程（空间，变量，文件信息等等的内容）也需要消耗时间。第二即便开启了成千上万的进程，操作系统也不能让他们同时执行，维护一个很大的进程列表的同时，调度的时候，还需要进行切换并且记录每个进程的执行节点，也就是记录上下文（各种变量等等乱七八糟的东西，虽然你看不到，但是操作系统都要做），这样反而会影响程序的效率。因此我们不能无限制的根据任务开启或者结束进程。就看我们上面的一些代码例子，你会发现有些程序是不是执行的时候比较慢才出结果，就是这个原因，那么我们要怎么做呢？
在这里，要给大家介绍一个进程池的概念，定义一个池子，在里面放上固定数量的进程，有需求来了，就拿一个池中的进程来处理任务，等到处理完毕，进程并不关闭，而是将进程再放回进程池中继续等待任务。如果有很多任务需要执行，池中的进程数量不够，任务就要等待之前的进程执行任务完毕归来，拿到空闲进程才能继续执行。也就是说，池中进程的数量是固定的，那么同一时间最多有固定数量的进程在运行。这样不会增加操作系统的调度难度，还节省了开闭进程的时间，也一定程度上能够实现并发效果
multiprocess.Poll模块
创建进程池的类：如果指定numprocess为3，则进程池会从无到有创建三个进程，然后自始至终使用这三个进程去执行所有任务（高级一些的进程池可以根据你的并发量，搞成动态增加或减少进程池中的进程数量的操作），不会开启其他进程，提高操作系统效率，减少空间的占用等。
概念介绍：
Pool([numprocess [,initializer [, initargs]]]):创建进程池
numprocess:要创建的进程数，如果省略，将默认使用cpu_count()的值
initializer：是每个工作进程启动时要执行的可调用对象，默认为None
initargs：是要传给initializer的参数组
p.apply(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。 ‘‘‘需要强调的是：此操作并不会在所有池工作进程中并执行func函数。如果要通过不同参数并发地执行func函数，必须从不同线程调用p.apply()函数或者使用p.apply_async()‘‘‘ p.apply_async(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。 ‘‘‘此方法的结果是AsyncResult类的实例，callback是可调用对象，接收输入参数。当func的结果变为可用时，将理解传递给callback。callback禁止执行任何阻塞操作，否则将接收其他异步操作中的结果。‘‘‘ p.close():关闭进程池，防止进一步操作。如果所有操作持续挂起，它们将在工作进程终止前完成 P.jion():等待所有工作进程退出。此方法只能在close（）或teminate()之后调用
方法apply_async()和map_async（）的返回值是AsyncResul的实例obj。实例具有以下方法
obj.get():返回结果，如果有必要则等待结果到达。timeout是可选的。如果在指定时间内还没有到达，将引发一场。如果远程操作中引发了异常，它将在调用此方法时再次被引发。
obj.ready():如果调用完成，返回True
obj.successful():如果调用完成且没有引发异常，返回True，如果在结果就绪之前调用此方法，引发异常
obj.wait([timeout]):等待结果变为可用。
obj.terminate()：立即终止所有工作进程，同时不执行任何清理或结束任何挂起工作。如果p被垃圾回收，将自动调用此函数
import time from multiprocessing import Pool,Process #针对range(100)这种参数的 # def func(n): # for i in range(3): # print(n + 1) def func(n): print(n) # 结果： # (1, 2) # alex def func2(n): for i in range(3): print(n - 1) if __name__ == ‘__main__‘: #1.进程池的模式 s1 = time.time() #我们计算一下开多进程和进程池的执行效率 poll = Pool(5) #创建含有5个进程的进程池 # poll.map(func,range(100)) #异步调用进程，开启100个任务,map自带join的功能 poll.map(func,[(1,2),‘alex‘]) #异步调用进程，开启100个任务,map自带join的功能 # poll.map(func2,range(100)) #如果想让进程池完成不同的任务，可以直接这样搞 #map只限于接收一个可迭代的数据类型参数，列表啊，元祖啊等等，如果想做其他的参数之类的操作，需要用后面我们要学的方法。 # t1 = time.time() - s1 # # #2.多进程的模式 # s2 = time.time() # p_list = [] # for i in range(100): # p = Process(target=func,args=(i,)) # p_list.append(p) # p.start() # [pp.join() for pp in p_list] # t2 = time.time() - s2 # # print(‘t1>>‘,t1) #结果：0.5146853923797607s 进程池的效率高 # print(‘t2>>‘,t2) #结果：12.092015027999878s
有一点，map是异步执行的，并且自带close和join
一般约定俗成的是进程池中的进程数量为CPU的数量，工作中要看具体情况来考量。
实际应用代码示例：
同步与异步两种执行方式：
import os,time from multiprocessing import Pool def work(n): print(‘%s run‘ %os.getpid()) time.sleep(1) return n**2 if __name__ == ‘__main__‘: p=Pool(3) #进程池中从无到有创建三个进程,以后一直是这三个进程在执行任务 res_l=[] for i in range(10): res=p.apply(work,args=(i,)) # 同步调用，直到本次任务执行完毕拿到res，等待任务work执行的过程中可能有阻塞也可能没有阻塞 # 但不管该任务是否存在阻塞，同步调用都会在原地等着 res_l.append(res) print(res_l)
import os import time import random from multiprocessing import Pool def work(n): print(‘%s run‘ %os.getpid()) time.sleep(random.random()) return n**2 if __name__ == ‘__main__‘: p=Pool(3) #进程池中从无到有创建三个进程,以后一直是这三个进程在执行任务 res_l=[] for i in range(10): res=p.apply_async(work,args=(i,)) # 异步运行，根据进程池中有的进程数，每次最多3个子进程在异步执行，并且可以执行不同的任务，传送任意的参数了。 # 返回结果之后，将结果放入列表，归还进程，之后再执行新的任务 # 需要注意的是，进程池中的三个进程不会同时开启或者同时结束 # 而是执行完一个就释放一个进程，这个进程就去接收新的任务。 res_l.append(res) # 异步apply_async用法：如果使用异步提交的任务，主进程需要使用join，等待进程池内任务都处理完，然后可以用get收集结果 # 否则，主进程结束，进程池可能还没来得及执行，也就跟着一起结束了 p.close() #不是关闭进程池，而是结束进程池接收任务，确保没有新任务再提交过来。 p.join() #感知进程池中的任务已经执行结束，只有当没有新的任务添加进来的时候，才能感知到任务结束了，所以在join之前必须加上close方法 for res in res_l: print(res.get()) #使用get来获取apply_aync的结果,如果是apply,则没有get方法,因为apply是同步执行,立刻获取结果,也根本无需get
进程池版的socket并发聊天代码示例：
#Pool内的进程数默认是cpu核数，假设为4（查看方法os.cpu_count()） #开启6个客户端，会发现2个客户端处于等待状态 #在每个进程内查看pid，会发现pid使用为4个，即多个客户端公用4个进程 from socket import * from multiprocessing import Pool import os server=socket(AF_INET,SOCK_STREAM) server.setsockopt(SOL_SOCKET,SO_REUSEADDR,1) server.bind((‘127.0.0.1‘,8080)) server.listen(5) def talk(conn): print(‘进程pid: %s‘ %os.getpid()) while True: try: msg=conn.recv(1024) if not msg:break conn.send(msg.upper()) except Exception: break if __name__ == ‘__main__‘: p=Pool(4) while True: conn,*_=server.accept() p.apply_async(talk,args=(conn,)) # p.apply(talk,args=(conn,client_addr)) #同步的话，则同一时间只有一个客户端能访问
from socket import * client=socket(AF_INET,SOCK_STREAM) client.connect((‘127.0.0.1‘,8080)) while True: msg=input(‘>>: ‘).strip() if not msg:continue client.send(msg.encode(‘utf-8‘)) msg=client.recv(1024) print(msg.decode(‘utf-8‘))
然后我们再提一个回调函数
需要回调函数的场景：进程池中任何一个任务一旦处理完了，就立即告知主进程：我好了额，你可以处理我的结果了。主进程则调用一个函数去处理该结果，该函数即回调函数，这是进程池特有的，普通进程没有这个机制，但是我们也可以通过进程通信来拿到返回值，进程池的这个回调也是进程通信的机制完成的。
我们可以把耗时间（阻塞）的任务放到进程池中，然后指定回调函数（主进程负责执行），这样主进程在执行回调函数时就省去了I/O的过程，直接拿到的是任务的结果
import os from multiprocessing import Pool def func1(n): print(‘func1>>‘,os.getpid()) print(‘func1‘) return n*n def func2(nn): print(‘func2>>‘,os.getpid()) print(‘func2‘) print(nn) # import time # time.sleep(0.5) if __name__ == ‘__main__‘: print(‘主进程：‘,os.getpid()) p = Pool(5) #args里面的10给了func1，func1的返回值作为回调函数的参数给了callback对应的函数，不能直接给回调函数直接传参数，他只能是你任务函数func1的函数的返回值 # for i in range(10,20): #如果是多个进程来执行任务，那么当所有子进程将结果给了回调函数之后，回调函数又是在主进程上执行的，那么就会出现打印结果是同步的效果。我们上面func2里面注销的时间模块打开看看 # p.apply_async(func1,args=(i,),callback=func2) p.apply_async(func1,args=(10,),callback=func2) p.close() p.join() #结果 # 主进程： 11852 #发现回调函数是在主进程中完成的，其实如果是在子进程中完成的，那我们直接将代码写在子进程的任务函数func1里面就行了，对不对，这也是为什么称为回调函数的原因。 # func1>> 17332 # func1 # func2>> 11852 # func2 # 100
回调函数在写的时候注意一点，回调函数的形参执行有一个，如果你的执行函数有多个返回值，那么也可以被回调函数的这一个形参接收，接收的是一个元祖，包含着你执行函数的所有返回值。
使用进程池来搞爬虫的时候，最耗时间的是请求地址的网络请求延迟，那么如果我们在将处理数据的操作加到每个子进程中，那么所有在进程池后面排队的进程就需要等更长的时间才能获取进程池里面的执行进程来执行自己，所以一般我们就将请求作成一个执行函数，通过进程池去异步执行，剩下的数据处理的内容放到另外一个进程或者主进程中去执行，将网络延迟的时间也利用起来，效率更高。
requests这个模块的get方法请求页面，就和我们在浏览器上输入一个网址然后回车去请求别人的网站的效果是一样的。安装requests模块的指令：在cmd窗口执行pip install requests。
进程池和信号量的区别：
进程池是多个需要被执行的任务在进程池外面排队等待获取进程对象去执行自己，而信号量是一堆进程等待着去执行一段逻辑代码。
信号量不能控制创建多少个进程，但是可以控制同时多少个进程能够执行，但是进程池能控制你可以创建多少个进程。
举例：就像那些开大车拉煤的，信号量是什么呢，就好比我只有五个车道，你每次只能过5辆车，但是不影响你创建100辆车，但是进程池相当于什么呢？相当于你只有5辆车，每次5个车拉东西，拉完你再把车放回来，给别的人拉煤用。
其他语言里面有更高级的进程池，在设置的时候，可以将进程池中的进程动态的创建出来，当需求增大的时候，就会自动在进程池中添加进程，需求小的时候，自动减少进程，并且可以设置进程数量的上线，最多为多，python里面没有。
以上是关于多进程相关的主要内容，如果未能解决你的问题，请参考以下文章