多道技术
多道技术中的多道指的是多个程序，多道技术的实现是为了解决多个程序竞争或者说共享同一个资源（比如cpu）的有序调度问题，解决方式即多路复用，多路复用分为时间上的复用和空间上的复用。
空间上的复用：将内存分为几部分，每个部分放入一个程序，这样，同一时间内存中就有了多道程序。
时间上的复用：当一个程序在等待I/O时，另一个程序可以使用cpu，如果内存中可以同时存放足够多的作业，则cpu的利用率可以接近100%，类似于我们小学数学所学的统筹方法。（操作系统采用了多道技术后，可以控制进程的切换，或者说进程之间去争抢cpu的执行权限。这种切换不仅会在一个进程遇到io时进行，一个进程占用cpu时间过长也会切换，或者说被操作系统夺走cpu的执行权限）
现代计算机或者网络都是多用户的，多个用户不仅共享硬件，而且共享文件，数据库等信息，共享意味着冲突和无序。
操作系统主要使用来干什么？
1.记录哪个程序使用什么资源 2.对资源请求进行分配 3.为不同的程序和用户调解互相冲突的资源请求。
我们可将上述操作系统的功能总结为：处理来自多个程序发起的多个（多个即多路）共享（共享即复用）资源的请求，简称多路复用多路复用有两种实现方式
时间上的复用
当一个资源在时间上复用时，不同的程序或用户轮流使用它，第一个程序获取 该资源使用结束后，在轮到第二个。。。第三个。。。 例如：只有一个cpu，多个程序需要在该cpu上运行，操作系统先把cpu分给第 一个程序，在这个程序运行的足够长的时间（时间长短由操作系统的算法说了算） 或者遇到了I/O阻塞，操作系统则把cpu分配给下一个程序，以此类推，直到第一个 程序重新被分配到了cpu然后再次运行，由于cpu的切换速度很快，给用户的感觉就 是这些程序是同时运行的，或者说是并发的，或者说是伪并行的。至于资源如何实现 时间复用，或者说谁应该是下一个要运行的程序，以及一个任务需要运行多长时间， 这些都是操作系统的工作。
空间上的复用
每个客户都获取了一个大的资源中的一小部分资源，从而减少了排队等待资源的时间。
例如：多个运行的程序同时进入内存，硬件层面提供保护机制来确保各自的内存是分
割开的，且由操作系统控制，这比一个程序独占内存一个一个排队进入内存效率要高的多。
有关空间复用的其他资源还有磁盘，在许多系统中，一个磁盘同时为许多用户保存文件。分配磁盘空间并且记录谁正在使用哪个磁盘块是操作系统资源管理的典型任务。这两种方式合起来便是多道技术详解
空间上的复用最大的问题是：
程序之间的内存必须分割，这种分割需要在硬件层面实现，由操作系统控制。如果内存彼
此不分割，则一个程序可以访问另外一个程序的内存，
首先丧失的是安全性，比如你的qq程序可以访问操作系统的内存，这意味着你的qq可以拿
到操作系统的所有权限。
其次丧失的是稳定性，某个程序崩溃时有可能把别的程序的内存也给回收了，比方说把操作
系统的内存给回收了，则操作系统崩溃。
什么是进程
进程：正在进行的一个过程或者说一个任务。而负责执行任务则是cpu。
并发与并行
无论是并行还是并发，在用户看来都是'同时'运行的，不管是进程还是线程，都只是一个任务而已，真实干活的是cpu，cpu来做这些任务，而一个cpu同一时刻只能执行一个任务
一 并发：是伪并行，即看起来是同时运行。单个cpu+多道技术就可以实现并发，（并行也属于并发）
二 并行：同时运行，只有具备多个cpu才能实现并行
单核下，可以利用多道技术，多个核，每个核也都可以利用多道技术（多道技术是针对单核而言的）
有四个核，六个任务，这样同一时间有四个任务被执行，假设分别被分配给了cpu1，cpu2，cpu3，cpu4，
一旦任务1遇到I/O就被迫中断执行，此时任务5就拿到cpu1的时间片去执行，这就是单核下的多道技术
而一旦任务1的I/O结束了，操作系统会重新调用它(需知进程的调度、分配给哪个cpu运行，由操作系统说了算)，可能被分配给四个cpu中的任意一个去执行
同步异步and阻塞非阻塞
状态介绍
在了解其他概念之前，我们首先要了解进程的几个状态。在程序运行的过程中，由于被操作系统的调度算法控制，程序会进入几个状态：就绪，运行和阻塞。 （1）就绪(Ready)状态 当进程已分配到除CPU以外的所有必要的资源，只要获得处理机便可立即执行，这时的进程状态称为就绪状态。 （2）执行/运行（Running）状态当进程已获得处理机，其程序正在处理机上执行，此时的进程状态称为执行状态。 （3）阻塞(Blocked)状态正在执行的进程，由于等待某个事件发生而无法执行时，便放弃处理机而处于阻塞状态。引起进程阻塞的事件可有多种，例如，等待I/O完成、申请缓冲区不能满足、等待信件(信号)等。
同步和异步关注的是消息通信机制.
同步异步指的是调用者与被调用者两者之间的关系，而不是经常容易误解的多个被调用者之间的关系
同步
所谓同步，就是在发出一个功能调用时，在没有得到结果之前，该调用就不会返回。按照这个定义，其实绝大多数函数都是同步调用。但是一般而言，我们在说同步、异步的时候，特指那些需要其他部件协作或者需要一定时间完成的任务。 举例： 1. multiprocessing.Pool下的apply #发起同步调用后，就在原地等着任务结束，根本不考虑任务是在计算还是在io阻塞，总之就是一股脑地等任务结束 2. concurrent.futures.ProcessPoolExecutor().submit(func,).result() 3. concurrent.futures.ThreadPoolExecutor().submit(func,).result()
异步
异步的概念和同步相对。当一个异步功能调用发出后，调用者不能立刻得到结果。当该异步功能完成后，通过状态、通知或回调来通知调用者。如果异步功能用状态来通知，那么调用者就需要每隔一定时间检查一次，效率就很低（有些初学多线程编程的人，总喜欢用一个循环去检查某个变量的值，这其实是一 种很严重的错误）。如果是使用通知的方式，效率则很高，因为异步功能几乎不需要做额外的操作。至于回调函数，其实和通知没太多区别。 举例： 1. multiprocessing.Pool().apply_async() #发起异步调用后，并不会等待任务结束才返回，相反，会立即获取一个临时结果（并不是最终的结果，可能是封装好的一个对象）。 2. concurrent.futures.ProcessPoolExecutor(3).submit(func,) 3. concurrent.futures.ThreadPoolExecutor(3).submit(func,)
阻塞
阻塞调用是指调用结果返回之前，当前线程会被挂起（如遇到io操作）。函数只有在得到结果之后才会将阻塞的线程激活。有人也许会把阻塞调用和同步调用等同起来，实际上他是不同的。对于同步调用来说，很多时候当前线程还是激活的，只是从逻辑上当前函数没有返回而已。 举例： 1. 同步调用：apply一个累计1亿次的任务，该调用会一直等待，直到任务返回结果为止，但并未阻塞住（即便是被抢走cpu的执行权限，那也是处于就绪态）; 2. 阻塞调用：当socket工作在阻塞模式的时候，如果没有数据的情况下调用recv函数，则当前线程就会被挂起，直到有数据为止。
非阻塞
#非阻塞和阻塞的概念相对应，指在不能立刻得到结果之前也会立刻返回，同时该函数不会阻塞当前线程。
小结
1. 同步与异步针对的是函数/任务的调用方式：同步就是当一个进程发起一个函数（任务）调用的时候，一直等到函数（任务）完成，而进程继续处于激活状态。而异步情况下是当一个进程发起一个函数（任务）调用的时候，不会等函数返回，而是继续往下执行当，函数返回的时候通过状态、通知、事件等方式通知进程任务完成。 2. 阻塞与非阻塞针对的是进程或线程：阻塞是当请求不能满足的时候就将进程挂起，而非阻塞则不会阻塞当前进程
补充
- 同步阻塞形式
效率最低。拿上面的例子来说，就是你专心排队，什么别的事都不做。
- 异步阻塞形式
如果在银行等待办理业务的人采用的是异步的方式去等待消息被触发（通知），也就是领了一张小纸条，假如在这段时间里他不能离开银行做其它的事情，那么很显然，这个人被阻塞在了这个等待的操作上面；
异步操作是可以被阻塞住的，只不过它不是在处理消息时阻塞，而是在等待消息通知时被阻塞。
- 同步非阻塞形式
实际上是效率低下的。
想象一下你一边打着电话一边还需要抬头看到底队伍排到你了没有，如果把打电话和观察排队的位置看成是程序的两个操作的话，这个程序需要在这两种不同的行为之间来回的切换，效率可想而知是低下的。
- 异步非阻塞形式
效率更高，
因为打电话是你(等待者)的事情，而通知你则是柜台(消息触发机制)的事情，程序没有在两种不同的操作中来回切换。
比如说，这个人突然发觉自己烟瘾犯了，需要出去抽根烟，于是他告诉大堂经理说，排到我这个号码的时候麻烦到外面通知我一下，那么他就没有被阻塞在这个等待的操作上面，自然这个就是异步+非阻塞的方式了。
很多人会把同步和阻塞混淆，是因为很多时候同步操作会以阻塞的形式表现出来，同样的，很多人也会把异步和非阻塞混淆，因为异步操作一般都不会在真正的IO操作处被阻塞。
进程的创建
但凡是硬件，都需要有操作系统去管理，只要有操作系统，就有进程的概念，就需要有创建进程的方式，一些操作系统只为一个应用程序设计，比如微波炉中的控制器，一旦启动微波炉，所有的进程都已经存在。
而对于通用系统（跑很多应用程序），需要有系统运行过程中创建或撤销进程的能力，主要分为4种形式创建新的进程
1. 系统初始化（查看进程linux中用ps命令，windows中用任务管理器，前台进程负责与用户交互，后台运行的进程与用户无关，运行在后台并且只在需要时才唤醒的进程，称为守护进程，如电子邮件、web页面、新闻、打印）
2. 一个进程在运行过程中开启了子进程（如nginx开启多进程，os.fork,subprocess.Popen等）
3. 用户的交互式请求，而创建一个新进程（如用户双击暴风影音）
4. 一个批处理作业的初始化（只在大型机的批处理系统中应用）
无论哪一种，新进程的创建都是由一个已经存在的进程执行了一个用于创建进程的系统调用而创建的：
1. 在UNIX中该系统调用是：fork，fork会创建一个与父进程一模一样的副本，二者有相同的存储映像、同样的环境字符串和同样的打开文件（在shell解释器进程中，执行一个命令就会创建一个子进程）
2. 在windows中该系统调用是：CreateProcess，CreateProcess既处理进程的创建，也负责把正确的程序装入新进程。
关于创建的子进程，UNIX和windows
1.相同的是：进程创建后，父进程和子进程有各自不同的地址空间（多道技术要求物理层面实现进程之间内存的隔离），任何一个进程的在其地址空间中的修改都不会影响到另外一个进程。
2.不同的是：在UNIX中，子进程的初始地址空间是父进程的一个副本，提示：子进程和父进程是可以有只读的共享内存区的。但是对于windows系统来说，从一开始父进程与子进程的地址空间就是不同的。
进程的终止
1. 正常退出（自愿，如用户点击交互式页面的叉号，或程序执行完毕调用发起系统调用正常退出，在linux中用exit，在windows中用ExitProcess）
2. 出错退出（自愿，python a.py中a.py不存在）
3. 严重错误（非自愿，执行非法指令，如引用不存在的内存，1/0等，可以捕捉异常，try...except...）
4. 被其他进程杀死（非自愿，如kill -9）
进程的层次结构
无论UNIX还是windows，进程只有一个父进程，不同的是：
1. 在UNIX中所有的进程，都是以init进程为根，组成树形结构。父子进程共同组成一个进程组，这样，当从键盘发出一个信号时，该信号被送给当前与键盘相关的进程组中的所有成员。
2. 在windows中，没有进程层次的概念，所有的进程都是地位相同的，唯一类似于进程层次的暗示，是在创建进程时，父进程得到一个特别的令牌（称为句柄）,该句柄可以用来控制子进程，但是父进程有权把该句柄传给其他子进程，这样就没有层次了。
进程的状态
tail -f access.log |grep '404'
执行程序tail，开启一个子进程，执行程序grep，开启另外一个子进程，两个进程之间基于管道'|'通讯，将tail的结果作为grep的输入。
进程grep在等待输入（即I/O）时的状态称为阻塞，此时grep命令都无法运行
其实在两种情况下会导致一个进程在逻辑上不能运行，
1. 进程挂起是自身原因，遇到I/O阻塞，便要让出CPU让其他进程去执行，这样保证CPU一直在工作
2. 与进程无关，是操作系统层面，可能会因为一个进程占用时间过多，或者优先级等原因，而调用其他的进程去使用CPU。
因而一个进程由三种状态
进程并发的实现
进程并发的实现在于，硬件中断一个正在运行的进程，把此时进程运行的所有状态保存下来，为此，操作系统维护一张表格，即进程表（process table），每个进程占用一个进程表项（这些表项也称为进程控制块）
该表存放了进程状态的重要信息：程序计数器、堆栈指针、内存分配状况、所有打开文件的状态、帐号和调度信息，以及其他在进程由运行态转为就绪态或阻塞态时，必须保存的信息，从而保证该进程在再次启动时，就像从未被中断过一样。
multiprocessing模块介绍
python中的多线程无法利用多核优势，如果想要充分地使用多核CPU的资源（os.cpu_count()查看），在python中大部分情况需要使用多进程。Python提供了multiprocessing。
multiprocessing模块用来开启子进程，并在子进程中执行我们定制的任务（比如函数），该模块与多线程模块threading的编程接口类似。
multiprocessing模块的功能众多：支持子进程、通信和共享数据、执行不同形式的同步，提供了Process、Queue、Pipe、Lock等组件。
需要再次强调的一点是：与线程不同，进程没有任何共享状态，进程修改的数据，改动仅限于该进程内。
Process类的介绍
创建进程的类：
Process([group [, target [, name [, args [, kwargs]]]]])，由该类实例化得到的对象， 可用来开启一个子进程 强调： 1. 需要使用关键字的方式来指定参数 2. args指定的为传给target函数的位置参数，是一个元组形式，必须有逗号
参数介绍：
group参数未使用，值始终为None target表示调用对象，即子进程要执行的任务 args表示调用对象的位置参数元组，args=(1,2,'egon',) kwargs表示调用对象的字典,kwargs={'name':'egon','age':18} name为子进程的名称
方法介绍：
p.start()：启动进程，并调用该子进程中的p.run()
p.run():进程启动时运行的方法，正是它去调用target指定的函数，我们自定义类的类中一定要实现该方法
p.terminate():强制终止进程p，不会进行任何清理操作，如果p创建了子进程，该子进程就成了僵尸进程，
使用该方法需要特别小心这种情况。如果p还保存了一个锁那么也将不会被释放，进而导致死锁
p.is_alive():如果p仍然运行，返回True
p.join([timeout]):主线程等待p终止（强调：是主线程处于等的状态，而p是处于运行的状态）。
timeout是可选的超时时间。
属性介绍：
p.daemon：默认值为False，如果设为True，代表p为后台运行的守护进程，当p的父进程
终止时，p也随之终止，并且设定为True后，p不能创建自己的新进程，必须在p.start()之前设置
p.name:进程的名称
p.pid：进程的pid
Process类的使用
windows下Process（）使用的解决方法及原理分析
import multiprocessing, time def test(i): while 1: print(i) time.sleep(.2) # if __name__ == '__main__': --- 去掉则会报错 multiprocessing.Process(target=test, args=(1,)).start() multiprocessing.Process(target=test, args=(2,)).start() #RuntimeError: # An attempt has been made to start a new process before the # current process has finished its bootstrapping phase. # This probably means that you are not using fork to start your # child processes and you have forgotten to use the proper idiom # in the main module: # if __name__ == '__main__': # freeze_support() # ... # The "freeze_support()" line can be omitted if the program # is not going to be frozen to produce an executable. 1234567891011121314151617181920212223242526 如果不加 if name == ‘main‘: 则会报错。 子进程会在运行时拷贝当前主进程中的所有内容，这也就意味着当一个新的子进程被创建的时候，该子进程就会复制当前模块，当然也包括了以下两行： multiprocessing.Process(target=test, args=(1,)).start() multiprocessing.Process(target=test, args=(2,)).start()12 很显然，这样的写法可能形成无限递归式地创建新的子进程。所以为了避免以上情况发生，我们在此引入了 if name == ‘main‘:
创建并开启子进程的方式一
import time import random from multiprocessing import Process def piao(name): print('%s piaoing' %name) time.sleep(random.randrange(1,5)) print('%s piao end' %name) if __name__ == '__main__': #实例化得到四个对象 p1=Process(target=piao,args=('egon',)) #必须加,号 p2=Process(target=piao,args=('alex',)) p3=Process(target=piao,args=('wupeqi',)) p4=Process(target=piao,args=('yuanhao',)) #调用对象下的方法，开启四个进程 p1.start() p2.start() p3.start() p4.start() print('主')
创建并开启子进程的方式二
import time import random from multiprocessing import Process class Piao(Process): def __init__(self,name): super().__init__() self.name=name def run(self): print('%s piaoing' %self.name) time.sleep(random.randrange(1,5)) print('%s piao end' %self.name) if __name__ == '__main__': #实例化得到四个对象 p1=Piao('egon') p2=Piao('alex') p3=Piao('wupeiqi') p4=Piao('yuanhao') #调用对象下的方法，开启四个进程 p1.start() #start会自动调用run p2.start() p3.start() p4.start() print('主')
主进程与子进程的执行顺序
from multiprocessing import Process import time def task(name): print("%s start" % name) time.sleep(3) print("%s stop" % name) if __name__ == '__main__': p = Process(target=task,args=("jerry",)) p.start() print("我是主进程!!!") time.sleep(100) print("我是主进程!!! over")
子进程创建需要时间，在这个空闲时间，父线程继续执行代码，子进程创建完成后显示。父进程与子进程是并行的。创建于启动所需要的时间，如果主进程不加sleep函数，足以跑完主进程，直到一个临界点才会进入执行子进程，会影响程序的输出顺序，并不会影响最后的结果
衍生出的两个问题
1.fork出一个子进程，父子进程执行的先后顺序是不确定的，如果先执行父进程，再执行子进程，父进程中没有wait和sleep。问，是否先把父进程执行完，再执行子进程？还是两个进程是一块执行的？
1. 进程的执行顺序是要看操作系统如何进行进程调度的，具体看调度算法。比如，如果基于时间片轮转调度算法，假设父进程先执行，本时间片内没有执行完，则会发生进程切换，即调度程序从就绪队列中取出一个进程在下一个时间片内执行，而本进程保存进程信息和状态然后插入就绪队列，等待调度程序重新调度。
2.如果父进程中有sleep，父进程中的程序执行到sleep进行休眠，转而执行子进程。
问:子进程中的程序执行完了再返回父进程中执行，还是休眠时间到了返回父进程中执行，还是其他？
sleep是休眠，也就是进程阻塞，从就绪队列取出本进程，插入阻塞队列。sleep时间到，进程变为就绪状态，插入就绪队列，等待调度程序调度，也就是说，执行不执行看调度程序，阻塞一定不执行，就绪也不一定是马上执行的。
进程调度
先来先服务（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队列的末尾，把处理机分配给新到的高优先权进程。
进程直接的内存空间是隔离的
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() print('主进程内：',n) 主进程内： 100 子进程： 0
僵尸进程与孤儿进程
参考博客：http://www.cnblogs.com/Anker/p/3271773.html 一：僵尸进程（有害） 僵尸进程：一个进程使用fork创建子进程，如果子进程退出，而父进程并没有调用wait或waitpid获取子进程的状态信息，那么子进程的进程描述符仍然保存在系统中。这种进程称之为僵死进程。详解如下 我们知道在unix/linux中，正常情况下子进程是通过父进程创建的，子进程在创建新的进程。子进程的结束和父进程的运行是一个异步过程,即父进程永远无法预测子进程到底什么时候结束，如果子进程一结束就立刻回收其全部资源，那么在父进程内将无法获取子进程的状态信息。 因此，UNⅨ提供了一种机制可以保证父进程可以在任意时刻获取子进程结束时的状态信息： 1、在每个进程退出的时候，内核释放该进程所有的资源，包括打开的文件，占用的内存等。但是仍然为其保留一定的信息（包括进程号the process ID，退出状态the termination status of the process，运行时间the amount of CPU time taken by the process等） 2、直到父进程通过wait / waitpid来取时才释放. 但这样就导致了问题，如果进程不调用wait / waitpid的话，那么保留的那段信息就不会释放，其进程号就会一直被占用，但是系统所能使用的进程号是有限的，如果大量的产生僵死进程，将因为没有可用的进程号而导致系统不能产生新的进程. 此即为僵尸进程的危害，应当避免。 任何一个子进程(init除外)在exit()之后，并非马上就消失掉，而是留下一个称为僵尸进程(Zombie)的数据结构，等待父进程处理。这是每个子进程在结束时都要经过的阶段。如果子进程在exit()之后，父进程没有来得及处理，这时用ps命令就能看到子进程的状态是“Z”。如果父进程能及时 处理，可能用ps命令就来不及看到子进程的僵尸状态，但这并不等于子进程不经过僵尸状态。 如果父进程在子进程结束之前退出，则子进程将由init接管。init将会以父进程的身份对僵尸状态的子进程进行处理。 二：孤儿进程（无害） 孤儿进程：一个父进程退出，而它的一个或多个子进程还在运行，那么那些子进程将成为孤儿进程。孤儿进程将被init进程(进程号为1)所收养，并由init进程对它们完成状态收集工作。 孤儿进程是没有父进程的进程，孤儿进程这个重任就落到了init进程身上，init进程就好像是一个民政局，专门负责处理孤儿进程的善后工作。每当出现一个孤儿进程的时候，内核就把孤 儿进程的父进程设置为init，而init进程会循环地wait()它的已经退出的子进程。这样，当一个孤儿进程凄凉地结束了其生命周期的时候，init进程就会代表党和政府出面处理它的一切善后工作。因此孤儿进程并不会有什么危害。 我们来测试一下（创建完子进程后，主进程所在的这个脚本就退出了，当父进程先于子进程结束时，子进程会被init收养，成为孤儿进程，而非僵尸进程），文件内容 import os import sys import time pid = os.getpid() ppid = os.getppid() print 'im father', 'pid', pid, 'ppid', ppid pid = os.fork() #执行pid=os.fork()则会生成一个子进程 #返回值pid有两种值： # 如果返回的pid值为0，表示在子进程当中 # 如果返回的pid值>0，表示在父进程当中 if pid > 0: print 'father died..' sys.exit(0) # 保证主线程退出完毕 time.sleep(1) print 'im child', os.getpid(), os.getppid() 执行文件，输出结果： im father pid 32515 ppid 32015 father died.. im child 32516 1 看，子进程已经被pid为1的init进程接收了，所以僵尸进程在这种情况下是不存在的，存在只有孤儿进程而已，孤儿进程声明周期结束自然会被init来销毁。 三：僵尸进程危害场景： 例如有个进程，它定期的产 生一个子进程，这个子进程需要做的事情很少，做完它该做的事情之后就退出了，因此这个子进程的生命周期很短，但是，父进程只管生成新的子进程，至于子进程 退出之后的事情，则一概不闻不问，这样，系统运行上一段时间之后，系统中就会存在很多的僵死进程，倘若用ps命令查看的话，就会看到很多状态为Z的进程。 严格地来说，僵死进程并不是问题的根源，罪魁祸首是产生出大量僵死进程的那个父进程。因此，当我们寻求如何消灭系统中大量的僵死进程时，答案就是把产生大 量僵死进程的那个元凶枪毙掉（也就是通过kill发送SIGTERM或者SIGKILL信号啦）。枪毙了元凶进程之后，它产生的僵死进程就变成了孤儿进 程，这些孤儿进程会被init进程接管，init进程会wait()这些孤儿进程，释放它们占用的系统进程表中的资源，这样，这些已经僵死的孤儿进程 就能瞑目而去了。 四：测试 #1、产生僵尸进程的程序test.py内容如下 #coding:utf-8 from multiprocessing import Process import time,os def run(): print('子',os.getpid()) if __name__ == '__main__': p=Process(target=run) p.start() print('主',os.getpid()) time.sleep(1000) #2、在unix或linux系统上执行 [root@vm172-31-0-19 ~]# python3 test.py & [1] 18652 [root@vm172-31-0-19 ~]# 主 18652 子 18653 [root@vm172-31-0-19 ~]# ps aux |grep Z USER PID %CPU %MEM VSZ RSS TTY STAT START TIME COMMAND root 18653 0.0 0.0 0 0 pts/0 Z 20:02 0:00 [python3] <defunct> #出现僵尸进程 root 18656 0.0 0.0 112648 952 pts/0 S+ 20:02 0:00 grep --color=auto Z [root@vm172-31-0-19 ~]# top #执行top命令发现1zombie top - 20:03:42 up 31 min, 3 users, load average: 0.01, 0.06, 0.12 Tasks: 93 total, 2 running, 90 sleeping, 0 stopped, 1 zombie %Cpu(s): 0.0 us, 0.3 sy, 0.0 ni, 99.7 id, 0.0 wa, 0.0 hi, 0.0 si, 0.0 st KiB Mem : 1016884 total, 97184 free, 70848 used, 848852 buff/cache KiB Swap: 0 total, 0 free, 0 used. 782540 avail Mem PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND root 20 0 29788 1256 988 S 0.3 0.1 0:01.50 elfin #3、 等待父进程正常结束后会调用wait／waitpid去回收僵尸进程 但如果父进程是一个死循环，永远不会结束，那么该僵尸进程就会一直存在，僵尸进程过多，就是有害的 解决方法一：杀死父进程 解决方法二：对开启的子进程应该记得使用join，join会回收僵尸进程 参考python2源码注释 class Process(object): def join(self, timeout=None): ''' Wait until child process terminates ''' assert self._parent_pid == os.getpid(), 'can only join a child process' assert self._popen is not None, 'can only join a started process' res = self._popen.wait(timeout) if res is not None: _current_process._children.discard(self) join方法中调用了wait，告诉系统释放僵尸进程。discard为从自己的children中剔除 解决方法三：http://blog.csdn.net/u010571844/article/details/50419798
查看进程与pid
class Piao(Process): def __init__(self, name): self.name=name super().__init__() #Process的__init__方法会执行self.name=Piao-1, #所以加到这里,会覆盖我们的self.name=name # 为我们开启的进程设置名字的正确做法 # super().__init__() # self.name = name def run(self): print('%s is piaoing' % self.name) time.sleep(random.randrange(1, 3)) print('%s is piao end' % self.name) if __name__ == '__main__': p = Piao('egon') p.start() print('开始') print(p.pid) # 查看pid # name与pid
父进程等待子进程结束
from multiprocessing import Process import time # def task(name): # print("%s start" % name) # time.sleep(3) # # print("%s stop" % name) # # if __name__ == '__main__': # p = Process(target=task,args=("jerry",)) # p.start() # p.join(3) #父进程必须等到子进程完成后才能继续执行 可设置等待超时时间 # print("我是主进程!!!") # def task(i): # print("%s start" % i) # # time.sleep(2) # # print("%s stop" % i) # # if __name__ == '__main__': # for i in range(1,11): # p = Process(target=task,args=(i,)) # p.start() # print("主进程!!!!") # 我们的代码只负责 通知操作系统创建进程 创建完就继续其他代码 # 但是操作系统什么时候创建完成 什么时候执行我们无法预知无法控制 def task(i): print("%s start" % i) time.sleep(2) print("%s stop" % i) if __name__ == '__main__': start_time = time.time() ps = [] for i in range(1,3): p = Process(target=task,args=(i,)) p.start() ps.append(p) # 主进程等子进程结束 for p in ps: p.join() print("主进程!!!!",time.time()-start_time)
Process对象的join方法
在主进程运行过程中如果想要并发的执行其他任务，我们可以开启子进程，此时主进程的任务和子进程的任务分为两种情况
一种情况是：在主进程的任务与子进程的任务彼此独立的情况下，主进程的任务先执行完毕后，主进程还需要等待子进程执行完毕，然后统一回收资源
一种情况是：如果主进程的任务在执行到某一个阶段时，需要等待子进程执行完毕后才能继续执行，就需要一种机制能够让主进程监测子进程是否运行完毕，在子进程执行完毕后才继续执行，否则一直在原地阻塞，这就是join方法的作用。
from multiprocessing import Process import time import os def task(name): print("%s is running ,parent is %s"%(name,os.getppid())) time.sleep(1) if __name__ == '__main__': p1 = Process(target=task,args=('子进程1',)) # print(p1.is_alive()) p2 = Process(target=task, args=('子进程2',)) p3 = Process(target=task, args=('子进程3',)) # print(p1.is_alive()) p1.start() print(p1.is_alive()) p2.start() p3.start() p1.join() p2.join() p3.join() print(p1.is_alive()) print("主进程 %s is running ,parent is %s" % (os.getpid(), os.getppid())) print(p1.name) True 子进程1 is running ,parent is 9360 子进程2 is running ,parent is 9360 子进程3 is running ,parent is 9360 False 主进程 9360 is running ,parent is 11764 Process-1 #没有自己传入名字就返回类名-1
有人会有疑问，既然join是等待进程结束，那么我像下面join（）下去，进程不就变成串行了的吗？
当然不是了，必须明确join是让谁等？
详细解析如下： 进程只要start就会在开始运行了,所以p1-p4.start()时,系统中已经有四个并发的进程了 而我们p1.join()是在等p1结束,没错p1只要不结束主线程就会一直卡在原地,这也是问题的关键 join是让主线程等,而p1-p4仍然是并发执行的,p1.join的时候,其余p2,p3,p4仍然在运行,等#p1.join结束,可能p2,p3,p4早已经结束了,这样p2.join,p3.join.p4.join直接通过检测，无需等待 所以4个join花费的总时间仍然是耗费时间最长的那个进程运行的时间
练习题：改写下面程序，分别实现下述打印效果
from multiprocessing import Process import time import random def task(n): time.sleep(random.randint(1,3)) print('-------->%s' %n) if __name__ == '__main__': p1=Process(target=task,args=(1,)) p2=Process(target=task,args=(2,)) p3=Process(target=task,args=(3,)) p1.start() p2.start() p3.start() print('-------->4')
效果一：保证最先输出-------->4
-------->4 -------->1 -------->3 -------->2
程序：不用修改
# _*_ coding: utf-8 _*_ from multiprocessing import Process import time import random def task(n): time.sleep(random.randint(1,2)) print("---------->%s"%n) if __name__ =='__main__': p1 =Process(target=task,args=(1,)) p2 = Process(target=task, args=(2,)) p3 = Process(target=task, args=(3,)) p1.start() p2.start() p3.start() print("---------->4")
效果二：保证最后输出-------->4
-------->2 -------->3 -------->1 -------->4
加上join
# _*_ coding: utf-8 _*_ from multiprocessing import Process import time import random def task(n): time.sleep(random.randint(1,2)) print("---------->%s"%n) if __name__ =='__main__': p1 =Process(target=task,args=(1,)) p2 = Process(target=task, args=(2,)) p3 = Process(target=task, args=(3,)) p1.start() p2.start() p3.start() p1.join() p2.join() p3.join() print("---------->4")
效果三：保证按顺序输出
-------->1 -------->2 -------->3 -------->4
# _*_ coding: utf-8 _*_ from multiprocessing import Process import time import random def task(n): time.sleep(random.randint(1,2)) print("---------->%s"%n) if __name__ =='__main__': p1 =Process(target=task,args=(1,)) p2 = Process(target=task, args=(2,)) p3 = Process(target=task, args=(3,)) '''这样写 是没有意义的，只是练习一下串行，这样写程序就成穿行的了 因为多进程的目的就是为了实现并发的效果''' p1.start() p1.join() p2.start() p2.join() p3.start() p3.join() print("---------->4")
思考题：判断上述三种效果，哪种属于并发，那种属于串行？
前两种属于并发，第三种属于串行
Process对象的其他属性或方法
进程对象的其他方法一:terminate与is_alive
from multiprocessing import Process import time import random def task(name): print('%s is piaoing' %name) time.sleep(random.randrange(1,5)) print('%s is piao end' %name) if __name__ == '__main__': p1=Process(target=task,args=('egon',)) p1.start() p1.terminate()#关闭进程,不会立即关闭,所以is_alive立刻查看的结果可能还是存活 print(p1.is_alive()) #结果为True print('主') print(p1.is_alive()) #结果为False
进程对象的其他属性：name与pid
from multiprocessing import Process import time import random def task(name): print('%s is piaoing' %name) time.sleep(random.randrange(1,5)) print('%s is piao end' %name) if __name__ == '__main__': p1=Process(target=task,args=('egon',),name='子进程1') #可以用关键参数来指定进程名 p1.start() print(p1.name,p1.pid,)
守护进程
主进程创建守护进程
其一：守护进程会在主进程代码执行结束后就终止
其二：守护进程内无法再开启子进程,否则抛出异常：AssertionError: daemonic processes are not allowed to have children
注意：进程之间是互相独立的，主进程代码运行结束，守护进程随即终止
from multiprocessing import Process import time import random class Piao(Process): def __init__(self,name): super().__init__() self.name = name def run(self): print('%s is piaoing '%self.name) time.sleep(random.randrange(1,3)) print('%s is piao end '%self.name) p=Piao('egon') p.daemon=True #一定在p.start()前设置，设置p为守护进程，禁止p创建子进程，并且父进程 # 代码执行结束，p即终止运行 p.start() print('子进程 %s'%p.name) print('主程序 ') 子进程 egon 主程序
互斥锁
进程之间数据不共享，但是共享同一套文件系统，所以访问同一个文件，或者打印终端是没有问题的，但是带来的是竞争，竞争带来的结果是错乱，如下：
from multiprocessing import Process import time,os def work(): print('%s is running'%os.getpid()) time.sleep(2) print('%s is done'%os.getpid()) if __name__ == '__main__': for i in range(3): p=Process(target=work) p.start() 8852 is running 12080 is running 592 is running 8852 is done 12080 is done 592 is done 可以看出当三个子进程同时操作一个文件时，一定会导致文件的混乱，数据安全受到了威胁
如何控制，就是加锁处理。而互斥锁的意思就是互相排斥，如果把多个进程比喻为多个人，互斥锁的工作原理就是多个人都要去争抢同一个资源：卫生间，一个人抢到卫生间后上一把锁，其他人都要等着，等到这个完成任务后释放锁，其他人才有可能有一个抢到......所以互斥锁的原理，就是把并发改成穿行，降低了效率，但保证了数据安全不错乱
#由并发变成了串行,牺牲了运行效率,但避免了竞争 from multiprocessing import Process,Lock import os,time def work(lock): lock.acquire() #加锁 print('%s is running' %os.getpid()) time.sleep(2) print('%s is done' %os.getpid()) lock.release() #释放锁 if __name__ == '__main__': lock=Lock() for i in range(3): p=Process(target=work,args=(lock,)) p.start()
模拟抢票练习
多个进程共享同一文件，我们可以把文件当数据库，用多个进程模拟多个人执行抢票任务
#文件db.txt的内容为：{"count":1} #注意一定要用双引号，不然json无法识别 from multiprocessing import Process import time,json def search(name): dic=json.load(open('db.txt')) time.sleep(1) print(' 33[43m%s 查到剩余票数%s 33[0m' %(name,dic['count'])) def get(name): dic=json.load(open('db.txt')) time.sleep(1) #模拟读数据的网络延迟 if dic['count'] >0: dic['count']-=1 time.sleep(1) #模拟写数据的网络延迟 json.dump(dic,open('db.txt','w')) print(' 33[46m%s 购票成功 33[0m' %name) def task(name): search(name) get(name) if __name__ == '__main__': for i in range(10): #模拟并发10个客户端抢票 name='<路人%s>' %i p=Process(target=task,args=(name,)) p.start()
并发运行，效率高，但竞争写同一文件，数据写入错乱,只有一张票，卖成功给了10个人
<路人0> 查到剩余票数1 <路人1> 查到剩余票数1 <路人2> 查到剩余票数1 <路人3> 查到剩余票数1 <路人4> 查到剩余票数1 <路人5> 查到剩余票数1 <路人6> 查到剩余票数1 <路人7> 查到剩余票数1 <路人8> 查到剩余票数1 <路人9> 查到剩余票数1 <路人0> 购票成功 <路人4> 购票成功 <路人1> 购票成功 <路人5> 购票成功 <路人3> 购票成功 <路人7> 购票成功 <路人2> 购票成功 <路人6> 购票成功 <路人8> 购票成功 <路人9> 购票成功
加锁处理：购票行为由并发变成了串行，牺牲了运行效率，但保证了数据安全
#把文件db.txt的内容重置为：{"count":1} from multiprocessing import Process,Lock import time,json def search(name): dic=json.load(open('db.txt')) time.sleep(1) print(' 33[43m%s 查到剩余票数%s 33[0m' %(name,dic['count'])) def get(name): dic=json.load(open('db.txt')) time.sleep(1) #模拟读数据的网络延迟 if dic['count'] >0: dic['count']-=1 time.sleep(1) #模拟写数据的网络延迟 json.dump(dic,open('db.txt','w')) print(' 33[46m%s 购票成功 33[0m' %name) def task(name,lock): search(name) with lock: #相当于lock.acquire(),执行完自代码块自动执行lock.release() get(name) if __name__ == '__main__': lock=Lock() for i in range(10): #模拟并发10个客户端抢票 name='<路人%s>' %i p=Process(target=task,args=(name,lock)) p.start()
执行结果：
<路人0> 查到剩余票数1 <路人1> 查到剩余票数1 <路人2> 查到剩余票数1 <路人3> 查到剩余票数1 <路人4> 查到剩余票数1 <路人5> 查到剩余票数1 <路人6> 查到剩余票数1 <路人7> 查到剩余票数1 <路人8> 查到剩余票数1 <路人9> 查到剩余票数1 <路人0> 购票成功
互斥锁与join
使用join可以将并发变成串行，互斥锁的原理也是将并发变成穿行，那我们直接使用join就可以了啊，为何还要互斥锁，说到这里我赶紧试了一下
#把文件db.txt的内容重置为：{"count":1} from multiprocessing import Process,Lock import time,json def search(name): dic=json.load(open('db.txt')) print(' 33[43m%s 查到剩余票数%s 33[0m' %(name,dic['count'])) def get(name): dic=json.load(open('db.txt')) time.sleep(1) #模拟读数据的网络延迟 if dic['count'] >0: dic['count']-=1 time.sleep(1) #模拟写数据的网络延迟 json.dump(dic,open('db.txt','w')) print(' 33[46m%s 购票成功 33[0m' %name) def task(name,): search(name) get(name) if __name__ == '__main__': for i in range(10): name='<路人%s>' %i p=Process(target=task,args=(name,)) p.start() p.join()
执行结果
<路人0> 查到剩余票数1 <路人0> 购票成功 <路人1> 查到剩余票数0 <路人2> 查到剩余票数0 <路人3> 查到剩余票数0 <路人4> 查到剩余票数0 <路人5> 查到剩余票数0 <路人6> 查到剩余票数0 <路人7> 查到剩余票数0 <路人8> 查到剩余票数0 <路人9> 查到剩余票数0
发现使用join将并发改成串行，确实能保证数据安全，但问题是连查票操作也变成只能一个一个人去查了，很明显大家查票时应该是并发地去查询而无需考虑数据准确与否，此时join与互斥锁的区别就显而易见了，join是将一个任务整体串行，而互斥锁的好处则是可以将一个任务中的某一段代码串行，比如只让task函数中的get任务串行
def task(name,): search(name) # 并发执行 lock.acquire() get(name) #串行执行 lock.release()
总结
加锁可以保证多个进程修改同一块数据时，同一时间只能有一个任务可以进行修改，即串行地修改，没错，速度是慢了，但牺牲了速度却保证了数据安全。
虽然可以用文件共享数据实现进程间通信，但问题是：
1、效率低（共享数据基于文件，而文件是硬盘上的数据）
2、需要自己加锁处理
因此我们最好找寻一种解决方案能够兼顾：
1、效率高（多个进程共享一块内存的数据）
2、帮我们处理好锁问题。
这就是mutiprocessing模块为我们提供的基于消息的IPC通信机制：队列和管道。
队列和管道都是将数据存放于内存中，而队列又是基于（管道+锁）实现的，可以让我们从复杂的锁问题中解脱出来，因而队列才是进程间通信的最佳选择。
我们应该尽量避免使用共享数据，尽可能使用消息传递和队列，避免处理复杂的同步和锁问题，而且在进程数目增多时，往往可以获得更好的可获展性。
队列
队列的特点是先进先出，后进后出
进程彼此之间互相隔离，要实现进程间通信（IPC），multiprocessing模块支持两种形式：队列和管道，这两种方式都是使用消息传递的
创建队列的类（底层就是以管道和锁定的方式实现）：
Queue([maxsize]):创建共享的进程队列，Queue是多进程安全的队列，
可以使用Queue实现多进程之间的数据传递。
参数介绍：
maxsize是队列中允许最大项数，省略则无大小限制。 但需要明确： 1、队列内存放的是消息而非大数据 2、队列占用的是内存空间，因而maxsize即便是无大小限制也受限于内存大小
主要方法介绍：
q.put方法用以插入数据到队列中。
q.get方法可以从队列读取并且删除一个元素。
队列的使用：
from multiprocessing import Process,Queue q=Queue(3) #put ,get ,put_nowait,get_nowait,full,empty q.put(1) q.put(2) q.put(3) print(q.full()) #满了 # q.put(4) #再放就阻塞住了 print(q.get()) print(q.get()) print(q.get()) print(q.empty()) #空了 # print(q.get()) #再取就阻塞住了
生产者消费者模型
在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。该模式通过平衡生产线程和消费线程的工作能力来提高程序的整体处理数据的速度。
为什么要使用生产者和消费者模式
在线程世界里，生产者就是生产数据的线程，消费者就是消费数据的线程。在多线程开发当中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。
什么是生产者消费者模式
生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。
基于队列实现生产者消费者模型
from multiprocessing import Process,Queue import time,random,os def consumer(q): while True: res=q.get() time.sleep(random.randint(1,3)) print(' 33[45m%s 吃 %s 33[0m' %(os.getpid(),res)) def producer(q): for i in range(10): time.sleep(random.randint(1,3)) res='包子%s' %i q.put(res) print(' 33[44m%s 生产了 %s 33[0m' %(os.getpid(),res)) if __name__ == '__main__': q=Queue() #生产者们:即厨师们 p1=Process(target=producer,args=(q,)) #消费者们:即吃货们 c1=Process(target=consumer,args=(q,)) #开始 p1.start() c1.start() print('主')
生产者消费者模型总结 #程序中有两类角色 一类负责生产数据（生产者） 一类负责处理数据（消费者） 引入生产者消费者模型为了解决的问题是： 平衡生产者与消费者之间的工作能力，从而提高程序整体处理数据的速度 如何实现： 生产者<-->队列<——>消费者 #生产者消费者模型实现类程序的解耦和
此时的问题是主进程永远不会结束，原因是：生产者p在生产完后就结束了，但是消费者c在取空了q之后，则一直处于死循环中且卡在q.get()这一步。
解决方式无非是让生产者在生产完毕后，往队列中再发一个结束信号，这样消费者在接收到结束信号后就可以break出死循环
from multiprocessing import Process,Queue import time,random,os def consumer(q): while True: res=q.get() if res is None:break #收到结束信号则结束 time.sleep(random.randint(1,3)) print(' 33[45m%s 吃 %s 33[0m' %(os.getpid(),res)) def producer(q): for i in range(10): time.sleep(random.randint(1,3)) res='包子%s' %i q.put(res) print(' 33[44m%s 生产了 %s 33[0m' %(os.getpid(),res)) q.put(None) #发送结束信号 if __name__ == '__main__': q=Queue() #生产者们:即厨师们 p1=Process(target=producer,args=(q,)) #消费者们:即吃货们 c1=Process(target=consumer,args=(q,)) #开始 p1.start() c1.start() print('主')
注意：结束信号None，不一定要由生产者发，主进程里同样可以发，但主进程需要等生产者结束后才应该发送该信号
from multiprocessing import Process,Queue import time,random,os def consumer(q): while True: res=q.get() if res is None:break #收到结束信号则结束 time.sleep(random.randint(1,3)) print(' 33[45m%s 吃 %s 33[0m' %(os.getpid(),res)) def producer(q): for i in range(2): time.sleep(random.randint(1,3)) res='包子%s' %i q.put(res) print(' 33[44m%s 生产了 %s 33[0m' %(os.getpid(),res)) if __name__ == '__main__': q=Queue() #生产者们:即厨师们 p1=Process(target=producer,args=(q,)) #消费者们:即吃货们 c1=Process(target=consumer,args=(q,)) #开始 p1.start() c1.start() p1.join() q.put(None) #发送结束信号 print('主')
但上述解决方式，在有多个生产者和多个消费者时，我们则需要用一个很low的方式去解决
from multiprocessing import Process,Queue import time,random,os def consumer(q): while True: res=q.get() if res is None:break #收到结束信号则结束 time.sleep(random.randint(1,3)) print(' 33[45m%s 吃 %s 33[0m' %(os.getpid(),res)) def producer(name,q): for i in range(2): time.sleep(random.randint(1,3)) res='%s%s' %(name,i) q.put(res) print(' 33[44m%s 生产了 %s 33[0m' %(os.getpid(),res)) if __name__ == '__main__': q=Queue() #生产者们:即厨师们 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,)) #开始 p1.start() p2.start() p3.start() c1.start() p1.join() #必须保证生产者全部生产完毕,才应该发送结束信号 p2.join() p3.join() q.put(None) #有几个消费者就应该发送几次结束信号None q.put(None) #发送结束信号 print('主')
其实我们的思路无非是发送结束信号而已，有另外一种队列提供了这种机制
JoinableQueue([maxsize])：这就像是一个Queue对象，但队列允许项目的使用者通知生成者项目已经被成功处理。通知进程是使用共享的信号和条件变量来实现的。 参数介绍： maxsize是队列中允许最大项数，省略则无大小限制。 #方法介绍： JoinableQueue的实例p除了与Queue对象相同的方法之外还具有： q.task_done()：使用者使用此方法发出信号，表示q.get()的返回项目已经被处理。如果调用此方法的次数大于从队列中删除项目的数量，将引发ValueError异常 q.join():生产者调用此方法进行阻塞，直到队列中所有的项目均被处理。阻塞将持续到队列中的每个项目均调用q.task_done（）方法为止
from multiprocessing import Process,JoinableQueue import time,random,os def consumer(q): while True: res=q.get() time.sleep(random.randint(1,3)) 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)) res='%s%s' %(name,i) q.put(res) print(' 33[44m%s 生产了 %s 33[0m' %(os.getpid(),res)) q.join() 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 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也没有存在的价值了,应该随着主进程的结束而结束,所以设置成守护进程
管道
进程间通信（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 import time,os def consumer(p,name): left,right=p left.close() while True: try: baozi=right.recv() print('%s 收到包子:%s' %(name,baozi)) except EOFError: right.close() break def producer(seq,p): left,right=p right.close() for i in seq: left.send(i) # time.sleep(1) else: left.close() if __name__ == '__main__': left,right=Pipe() c1=Process(target=consumer,args=((left,right),'c1')) c1.start() seq=(i for i in range(10)) producer(seq,(left,right)) right.close() left.close() c1.join() print('主进程')
注意：生产者和消费者都没有使用管道的某个端点，就应该将其关闭，如在生产者中关闭管道的右端，在消费者中关闭管道的左端。如果忘记执行这些步骤，程序可能再消费者中的recv()操作上挂起。管道是由操作系统进行引用计数的,必须在所有进程中关闭管道后才能生产EOFError异常。因此在生产者中关闭管道不会有任何效果，付费消费者中也关闭了相同的管道端点。
文件描述符详解参考：https://blog.csdn.net/dangzhangjing97/article/details/79653448
管道可以用于双向通信，利用通常在客户端/服务器中使用的请求／响应模型或远程过程调用，就可以使用管道编写与进程交互的程序
from multiprocessing import Process,Pipe import time,os def adder(p,name): server,client=p client.close() while True: try: x,y=server.recv() except EOFError: server.close() break res=x+y server.send(res) print('server done') if __name__ == '__main__': server,client=Pipe() c1=Process(target=adder,args=((server,client),'c1')) c1.start() server.close() client.send((10,20)) print(client.recv()) client.close() c1.join() print('主进程') #注意：send()和recv()方法使用pickle模块对对象进行序列化。
进程间数据交互的第三种方式：Manager
Python实现多进程间通信的方式有很多种，例如队列，管道等。
但是这些方式只适用于多个进程都是源于同一个父进程的情况。
如果多个进程不是源于同一个父进程，只能用共享内存，信号量等方式，但是这些方式对于复杂的数据结构，例如Queue，dict，list等，使用起来比较麻烦，不够灵活。
Manager是一种较为高级的多进程通信方式，它能支持Python支持的的任何数据结构。
它的原理是：先启动一个ManagerServer进程，这个进程是阻塞的，它监听一个socket，然后其他进程（ManagerClient）通过socket来连接到ManagerServer，实现通信。
manager.py代码，实现server和client两个类
''' 进程间通信 ''' from multiprocessing.managers import BaseManager from multiprocessing import RLock MANAGER_PORT = 6000 MANAGER_DOMAIN = '0.0.0.0' MANAGER_AUTH_KEY = 'aaaaaaaaaaaaaaa' #定义一个Manager类 class InfoManager(BaseManager): pass class DictItem(): def __init__(self, ): self.items = dict() def set(self, key, value): self.items[key] = value def get(self, key): return self.items.get(key) def __setitem__(self, key, value): self.set(key, value) #为这个manager类注册存储容器，也就是通过这个manager类实现的共享的变量， #这个变量最好是一个类实例，自己定义的或者python自动的类的实例都可以 #这里不能把d改成dict()，因为Client那边执行d['keyi']='value'的时候会报错：d这个变量不能修改 d = DictItem() lock = RLock() InfoManager.register('dict', callable=lambda: d) InfoManager.register('open_qq_login_lock', callable=lambda: lock) class ManagerServer(): ''' multiprocess Manager服务类 ''' def __init__(self, domain, port, auth_key): self.domain = domain self.port = port self.auth_key = auth_key def start_manager_server(self): self.queue_manager = InfoManager(address=('', self.port), authkey=self.auth_key) # self.dict = self.queue_manager.dict() self.server = self.queue_manager.get_server() def run(self): self.start_manager_server() self.server.serve_forever() def stop(self): self.server.shutdown() self.is_stop = 1 class ManagerClient(): ''' 访问mutiprocess Manager的类 ''' def __init__(self, domain, port, auth_key): self.domain = domain self.port = port self.auth_key = auth_key # self.get_share_dict() self.info_manager = InfoManager(address=(self.domain, self.port), authkey=self.auth_key) self.info_manager.connect() def get_dict(self): # self.dict = m.dict() self.dict = self.info_manager.dict() return self.dict def get_open_qq_login_lock(self): self.open_qq_login_lock = self.info_manager.open_qq_login_lock() return self.open_qq_login_lock if __name__ == '__main__': pass
用法
1.启动一个ManagerServer，这个进程是阻塞的
import manager def run(): manager_server = manager.ManagerServer(manager.MANAGER_DOMAIN, manager.MANAGER_PORT, manager.MANAGER_AUTH_KEY) manager_server.run() if __name__ == '__main__': run()
2.实例化一个client，获取共享的变量
# 进程间共享变量 manager_client = manager.ManagerClient(manager.MANAGER_DOMAIN, manager.MANAGER_PORT, manager.MANAGER_AUTH_KEY) share_dict = manager_client.get_dict() open_qq_login_lock = manager_client.get_open_qq_login_lock()
注意：
1.对client获取的变量修改，不会影响server那边的变量，例如server中，client1获取变量params1，把它修改为params2，server那边的数据不会修改。如果要修改server的数据，需要调用server的方法，把server那边的数据刷新。
#Manager实现数据交互与共享 from multiprocessing import Process,Manager import os def run(d,l): d[os.getpid()]=os.getpid() #os.getpid() 获得进程ID l.append(os.getpid()) print(l) if __name__=='__main__': with Manager() as manager: D=manager.dict() #生成一个字典，可在多进程间共享和传递 L=manager.list() #生成一个列表，可在多进程间共享和传递 p_obj=[] for i in range(10): p=Process(target=run,args=(D,L,)) p.start() p_obj.append(p) for p in p_obj: p.join() print(D) print(L)
from multiprocessing import Process,Manager import time def task(dic): print("子进程xxxxx") # li[0] = 1 # print(li[0]) dic["name"] = "xx" if __name__ == '__main__': m = Manager() # li = m.list([100]) dic = m.dict({}) #dic = m.dict() {}加不加都行 # 开启子进程 p = Process(target=task,args=(dic,)) p.start() time.sleep(3) print(dic)
信号量
互斥锁 同时只允许一个线程更改数据，而Semaphore是同时允许一定数量的线程更改数据 ，比如厕所有3个坑，那最多只允许3个人上厕所
后面的人只能等里面有人出来了才能再进去，如果指定信号量为3，那么来一个人获得一把锁计数加1，当计数等于3时，后面的人均需要等待。一旦释放，就有人可以获得一把锁信号量与进程池的概念很像，但是要区分开，信号量涉及到加锁的概念
from multiprocessing import Process,Semaphore import time,random def go_wc(sem,user): sem.acquire() print('%s 占到一个茅坑' %user) time.sleep(random.randint(0,3)) #模拟每个人拉屎速度不一样，0代表有的人蹲下就起来了 sem.release() if __name__ == '__main__': sem=Semaphore(5) p_l=[] for i in range(13): p=Process(target=go_wc,args=(sem,'user%s' %i,)) p.start() p_l.append(p) for i in p_l: i.join() print('============》')
from threading import Thread,Semaphore,current_thread,active_count import time # 用于控制 同时执行被锁定代码的线程数量 也就是线程的并发数量 # 也是一种锁 sm = Semaphore(2) def task(): sm.acquire() for i in range(10): print(current_thread(),time.ctime()) time.sleep(1) sm.release() def task2(): for i in range(10): print(current_thread(),time.ctime()) time.sleep(0.5) for i in range(5): Thread(target=task).start() # Thread(target=task2).start() # print(active_count(),time.ctime()) 执行结果 每次都有两个线程同时生成 <Thread(Thread-1, started 1404)> Tue Nov 13 16:34:09 2018 <Thread(Thread-2, started 12728)> Tue Nov 13 16:34:09 2018 <Thread(Thread-1, started 1404)> Tue Nov 13 16:34:10 2018 <Thread(Thread-2, started 12728)> Tue Nov 13 16:34:10 2018 <Thread(Thread-1, started 1404)> Tue Nov 13 16:34:11 2018 <Thread(Thread-2, started 12728)> Tue Nov 13 16:34:11 2018 <Thread(Thread-1, started 1404)> Tue Nov 13 16:34:12 2018 <Thread(Thread-2, started 12728)> Tue Nov 13 16:34:12 2018 <Thread(Thread-1, started 1404)> Tue Nov 13 16:34:13 2018 <Thread(Thread-2, started 12728)> Tue Nov 13 16:34:13 2018 <Thread(Thread-1, started 1404)> Tue Nov 13 16:34:14 2018 <Thread(Thread-2, started 12728)> Tue Nov 13 16:34:14 2018 <Thread(Thread-1, started 1404)> Tue Nov 13 16:34:15 2018 <Thread(Thread-2, started 12728)> Tue Nov 13 16:34:15 2018 <Thread(Thread-2, started 12728)> Tue Nov 13 16:34:16 2018 <Thread(Thread-1, started 1404)> Tue Nov 13 16:34:16 2018 <Thread(Thread-2, started 12728)> Tue Nov 13 16:34:17 2018 <Thread(Thread-1, started 1404)> Tue Nov 13 16:34:17 2018 <Thread(Thread-2, started 12728)> Tue Nov 13 16:34:18 2018 <Thread(Thread-1, started 1404)> Tue Nov 13 16:34:18 2018 <Thread(Thread-3, started 12824)> Tue Nov 13 16:34:19 2018 <Thread(Thread-4, started 14928)> Tue Nov 13 16:34:20 2018 <Thread(Thread-3, started 12824)> Tue Nov 13 16:34:20 2018 <Thread(Thread-4, started 14928)> Tue Nov 13 16:34:21 2018 <Thread(Thread-3, started 12824)> Tue Nov 13 16:34:22 2018 <Thread(Thread-4, started 14928)> Tue Nov 13 16:34:22 2018 <Thread(Thread-3, started 12824)> Tue Nov 13 16:34:23 2018 <Thread(Thread-4, started 14928)> Tue Nov 13 16:34:23 2018 <Thread(Thread-3, started 12824)> Tue Nov 13 16:34:24 2018 <Thread(Thread-4, started 14928)> Tue Nov 13 16:34:24 2018 <Thread(Thread-3, started 12824)> Tue Nov 13 16:34:25 2018 <Thread(Thread-4, started 14928)> Tue Nov 13 16:34:25 2018 <Thread(Thread-3, started 12824)> Tue Nov 13 16:34:26 2018 <Thread(Thread-4, started 14928)> Tue Nov 13 16:34:26 2018 <Thread(Thread-3, started 12824)> Tue Nov 13 16:34:27 2018 <Thread(Thread-4, started 14928)> Tue Nov 13 16:34:27 2018 <Thread(Thread-3, started 12824)> Tue Nov 13 16:34:28 2018 <Thread(Thread-4, started 14928)> Tue Nov 13 16:34:28 2018 <Thread(Thread-3, started 12824)> Tue Nov 13 16:34:29 2018 <Thread(Thread-4, started 14928)> Tue Nov 13 16:34:29 2018 <Thread(Thread-5, started 13588)> Tue Nov 13 16:34:30 2018 <Thread(Thread-5, started 13588)> Tue Nov 13 16:34:31 2018 <Thread(Thread-5, started 13588)> Tue Nov 13 16:34:32 2018 <Thread(Thread-5, started 13588)> Tue Nov 13 16:34:33 2018 <Thread(Thread-5, started 13588)> Tue Nov 13 16:34:34 2018 <Thread(Thread-5, started 13588)> Tue Nov 13 16:34:35 2018 <Thread(Thread-5, started 13588)> Tue Nov 13 16:34:36 2018 <Thread(Thread-5, started 13588)> Tue Nov 13 16:34:37 2018 <Thread(Thread-5, started 13588)> Tue Nov 13 16:34:38 2018 <Thread(Thread-5, started 13588)> Tue Nov 13 16:34:39 2018
semaphore是一个内置的计数器
每当调用acquire()时，内置计数器-1
每当调用release()时，内置计数器+1
计数器不能小于0，当计数器为0时，acquire()将阻塞线程直到其他线程调用release()
import time import threading def foo(): time.sleep(2) #程序休息2秒 print("ok",time.ctime()) for i in range(20): t1=threading.Thread(target=foo,args=()) #实例化一个线程 t1.start() #启动线程 #执行结果 ''' ok Tue Nov 13 16:38:43 2018 ok Tue Nov 13 16:38:43 2018 ok Tue Nov 13 16:38:43 2018 ok Tue Nov 13 16:38:43 2018 ok Tue Nov 13 16:38:43 2018 ok Tue Nov 13 16:38:43 2018 ok Tue Nov 13 16:38:43 2018 ok Tue Nov 13 16:38:43 2018 ok Tue Nov 13 16:38:43 2018 ok Tue Nov 13 16:38:43 2018 ok Tue Nov 13 16:38:43 2018 ok Tue Nov 13 16:38:43 2018 ok Tue Nov 13 16:38:43 2018 ok Tue Nov 13 16:38:43 2018 ok Tue Nov 13 16:38:43 2018 ok Tue Nov 13 16:38:43 2018 ok Tue Nov 13 16:38:43 2018 ok Tue Nov 13 16:38:43 2018 ok Tue Nov 13 16:38:43 2018 ok Tue Nov 13 16:38:43 2018 '''
可以看到，程序会在很短的时间内生成20个线程来打印一句话。
如果在主机执行IO密集型任务的时候再执行这种类型的程序时，计算机就有很大可能会宕机。
这时候就可以为这段程序添加一个计数器功能，来限制一个时间点内的线程数量。
代码如下:
import time import threading s1=threading.Semaphore(5) #添加一个计时器 def foo(): s1.acquire() #计数器获得锁 time.sleep(2) #程序休眠2秒 print('ok',time.ctime()) s1.release() for i in range(20): t1=threading.Thread(target=foo,args=()) #创建线程 t1.start() #启动线程 ''' ok Tue Nov 13 16:44:06 2018 ok Tue Nov 13 16:44:06 2018 ok Tue Nov 13 16:44:06 2018 ok Tue Nov 13 16:44:06 2018 ok Tue Nov 13 16:44:06 2018 ok Tue Nov 13 16:44:08 2018 ok Tue Nov 13 16:44:08 2018 ok Tue Nov 13 16:44:08 2018 ok Tue Nov 13 16:44:08 2018 ok Tue Nov 13 16:44:08 2018 ok Tue Nov 13 16:44:10 2018 ok Tue Nov 13 16:44:10 2018 ok Tue Nov 13 16:44:10 2018 ok Tue Nov 13 16:44:10 2018 ok Tue Nov 13 16:44:10 2018 ok Tue Nov 13 16:44:12 2018 ok Tue Nov 13 16:44:12 2018 ok Tue Nov 13 16:44:12 2018 ok Tue Nov 13 16:44:12 2018 ok Tue Nov 13 16:44:12 2018 '''
进程池
在利用Python进行系统管理的时候，特别是同时操作多个文件目录，或者远程控制多台主机，并行操作可以节约大量的时间。多进程是实现并发的手段之一，需要注意的问题是：
很明显需要并发执行的任务通常要远大于核数一个操作系统不可能无限开启进程，通常有几个核就开几个进程进程开启过多，效率反而会下降（开启进程是需要占用系统资源的，而且开启多余核数目的进程也无法做到并行）
例如当被操作对象数目不大时，可以直接利用multiprocessing中的Process动态成生多个进程，十几个还好，但如果是上百个，上千个。。。手动的去限制进程数量却又太过繁琐，此时可以发挥进程池的功效。
我们就可以通过维护一个进程池来控制进程数目，比如httpd的进程模式，规定最小进程数和最大进程数...
ps：对于远程过程调用的高级应用程序而言，应该使用进程池，Pool可以提供指定数量的进程，供用户调用，当有新的请求提交到pool中时，如果池还没有满，那么就会创建一个新的进程用来执行该请求；但如果池中的进程数已经达到规定最大值，那么该请求就会等待，直到池中有进程结束，就重用进程池中的进程。
创建进程池的类：如果指定numprocess为3，则进程池会从无到有创建三个进程，然后自始至终使用这三个进程去执行所有任务，不会开启其他进程
1 Pool([numprocess [,initializer [, initargs]]]):创建进程池
参数介绍：
1 numprocess:要创建的进程数，如果省略，将默认使用cpu_count()的值 2 initializer：是每个工作进程启动时要执行的可调用对象，默认为None 3 initargs：是要传给initializer的参数组
方法介绍：
主要方法：
1.p.apply(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。 '''需要强调的是：此操作并不会在所有池工作进程中并执行func函数。如果要通过不同参数并发地执行func函数，必须从不同线程调用p.apply()函数或者使用2.p.apply_async()''' p.apply_async(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。 '''此方法的结果是AsyncResult类的实例，callback是可调用对象，接收输入参数。当func的结果变为可用时，将理解传递给callback。callback禁止执行任何阻塞操作，否则将接收其他异步操作中的结果。''' 3.p.close():关闭进程池，防止进一步操作。如果所有操作持续挂起，它们将在工作进程终止前完成 4.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被垃圾回收，将自动调用此函数
应用：
#同步调用一 from multiprocessing import Pool import os,time def work(n): print('%s run' %os.getpid(),time.ctime()) time.sleep(2) return n**2 if __name__ == '__main__': p=Pool(3) #进程池中从无到有创建三个进程,以后一直是这三个进程在执行任务 start = time.time() res_l=[] for i in range(5): res=p.apply(work,args=(i,)) #同步调用，直到本次任务执行完毕拿到res，等待任务work执行的过程中可能有阻塞也可能没有阻塞，但不管该任务是否存在阻塞，同步调用都会在原地等着，只是等的过程中若是任务发生了阻塞就会被夺走cpu的执行权限 res_l.append(res) print(res_l) stop = time.time() print(stop - start) ''' 3504 run Tue Nov 13 18:31:39 2018 10940 run Tue Nov 13 18:31:41 2018 848 run Tue Nov 13 18:31:43 2018 3504 run Tue Nov 13 18:31:45 2018 10940 run Tue Nov 13 18:31:47 2018 [0, 1, 4, 9, 16] 10.19660758972168 '''
# 同步调用二 from concurrent.futures import ProcessPoolExecutor import time, os def work(n): print('%s run' % os.getpid(),time.ctime()) time.sleep(2) return n ** 2 if __name__ == '__main__': p = ProcessPoolExecutor(3) start = time.time() res_l=[] for i in range(5): res=p.submit(work,i).result() #同步调用 res_l.append(res) p.shutdown(wait=True) print(res_l) stop = time.time() print(stop - start) ''' 6628 run Tue Nov 13 18:32:19 2018 2380 run Tue Nov 13 18:32:21 2018 7260 run Tue Nov 13 18:32:23 2018 6628 run Tue Nov 13 18:32:25 2018 2380 run Tue Nov 13 18:32:27 2018 [0, 1, 4, 9, 16] 10.312713623046875 '''
from multiprocessing import Pool import os,time def work(n): print('%s run' %os.getpid(),time.ctime()) time.sleep(3) return n**2 if __name__ == '__main__': p=Pool(3) #进程池中从无到有创建三个进程,以后一直是这三个进程在执行任务 start = time.time() res_l=[] for i in range(5): res=p.apply_async(work,args=(i,)) #异步运行,阻塞、直到本次任务执行完毕拿到res res_l.append(res) #异步apply_async用法：如果使用异步提交的任务，主进程需要使用jion，等待进程池内任务都处理完，然后可以用get收集结果，否则，主进程结束，进程池可能还没来得及执行，也就跟着一起结束了 p.close() p.join() # print(res_l) 列表对象中存储了五个对象的地址 for res in res_l: print(res.get()) #使用get来获取apply_aync的结果,如果是apply,则没有get方法,因为apply是同步执行,立刻获取结果,也根本无需get stop = time.time() print(stop - start) ''' 8020 run Tue Nov 13 19:08:39 2018 9856 run Tue Nov 13 19:08:39 2018 496 run Tue Nov 13 19:08:39 2018 8020 run Tue Nov 13 19:08:42 2018 9856 run Tue Nov 13 19:08:42 2018 0 1 4 9 16 6.426192998886108 '''
from concurrent.futures import ThreadPoolExecutor import os,time def work(n): print('%s'%os.getpid(),time.ctime()) time.sleep(2) return n**2 if __name__ == '__main__': p1 = ThreadPoolExecutor(3) start = time.time() res_l = [] for i in range(5): res=p1.submit(work, i) res_l.append(res) p1.shutdown(wait=True) print(res_l) ''' [<Future at 0x17a401d21d0 state=finished returned int>, <Future at 0x17a404c7da0 state=finished returned int>, <Future at 0x17a404de0b8 state=finished returned int>, <Future at 0x17a404de940 state=finished returned int>, <Future at 0x17a404de9b0 state=finished returned int>] ''' # for res in res_l: # print(res.result()) #我们可以用obj.result方法得到值 end = time.time() print(end - start) ''' 12088 Tue Nov 13 19:06:11 2018 12088 Tue Nov 13 19:06:11 2018 12088 Tue Nov 13 19:06:11 2018 12088 Tue Nov 13 19:06:13 2018 12088 Tue Nov 13 19:06:13 2018 [<Future at 0x17a401d21d0 state=finished returned int>, <Future at 0x17a404c7da0 state=finished returned int>, <Future at 0x17a404de0b8 state=finished returned int>, <Future at 0x17a404de940 state=finished returned int>, <Future at 0x17a404de9b0 state=finished returned int>] 4.003036737442017 '''
#一：使用进程池（异步调用,apply_async） #coding: utf-8 from multiprocessing import Process,Pool import time def func(msg): print( "msg:", msg) time.sleep(1) return msg if __name__ == "__main__": pool = Pool(processes = 3) res_l=[] for i in range(10): msg = "hello %d" %(i) res=pool.apply_async(func, (msg, )) #维持执行的进程总数为processes，当一个进程执行完毕后会添加新的进程进去 res_l.append(res) print("==============================>") #没有后面的join，或get，则程序整体结束，进程池中的任务还没来得及全部执行完也都跟着主进程一起结束了 pool.close() #关闭进程池，防止进一步操作。如果所有操作持续挂起，它们将在工作进程终止前完成 pool.join() #调用join之前，先调用close函数，否则会出错。执行完close后不会有新的进程加入到pool,join函数等待所有子进程结束 print(res_l) #看到的是<multiprocessing.pool.ApplyResult object at 0x10357c4e0>对象组成的列表,而非最终的结果,但这一步是在join后执行的,证明结果已经计算完毕,剩下的事情就是调用每个对象下的get方法去获取结果 for i in res_l: print(i.get()) #使用get来获取apply_aync的结果,如果是apply,则没有get方法,因为apply是同步执行,立刻获取结果,也根本无需get #二：使用进程池（同步调用,apply） #coding: utf-8 from multiprocessing import Process,Pool import time def func(msg): print( "msg:", msg) time.sleep(0.1) return msg if __name__ == "__main__": pool = Pool(processes = 3) res_l=[] for i in range(10): msg = "hello %d" %(i) res=pool.apply(func, (msg, )) #维持执行的进程总数为processes，当一个进程执行完毕后会添加新的进程进去 res_l.append(res) #同步执行，即执行完一个拿到结果，再去执行另外一个 print("==============================>") pool.close() pool.join() #调用join之前，先调用close函数，否则会出错。执行完close后不会有新的进程加入到pool,join函数等待所有子进程结束 print(res_l) #看到的就是最终的结果组成的列表 for i in res_l: #apply是同步的，所以直接得到结果，没有get()方法 print(i)
下一节会详细介绍 Python标准模块--concurrent.futures，比单纯的进程池和线程池好用
使用进程池维护固定数目的进程
''' 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,client_addr): 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() #当填3时，从服务端只能看到三个pid，当三个客户端其中一个客户端被杀死后才能连接上一个新的客户端 while True: conn,client_addr = server.accept() p.apply_async(talk,args=(conn,client_addr,)) # 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'))
发现：并发开启多个客户端，服务端同一时间只有4个不同的pid，干掉一个客户端，另外一个客户端才会进来，被4个进程之一处理
回调函数
需要回调函数的场景：进程池中任何一个任务一旦处理完了，就立即告知主进程：我好了额，你可以处理我的结果了。主进程则调用一个函数去处理该结果，该函数即回调函数
我们可以把耗时间（阻塞）的任务放到进程池中，然后指定回调函数（主进程负责执行），这样主进程在执行回调函数时就省去了I/O的过程，直接拿到的是任务的结果。
from multiprocessing import Pool import requests import json import os def get_page(url): print('<进程%s> get %s' %(os.getpid(),url)) respone=requests.get(url) if respone.status_code == 200: return {'url':url,'text':respone.text} def pasrse_page(res): print('<进程%s> parse %s' %(os.getpid(),res['url'])) parse_res='url:<%s> size:[%s] ' %(res['url'],len(res['text'])) with open('db.txt','a') as f: f.write(parse_res) if __name__ == '__main__': urls=[ 'https://www.baidu.com', 'https://www.python.org', 'https://www.openstack.org', 'https://help.github.com/', 'http://www.sina.com.cn/' ] p=Pool(3) res_l=[] for url in urls: res=p.apply_async(get_page,args=(url,),callback=pasrse_page) res_l.append(res) p.close() p.join() print([res.get() for res in res_l]) #拿到的是get_page的结果,其实完全没必要拿该结果,该结果已经传给回调函数处理了 ''' 打印结果: <进程3388> get https://www.baidu.com <进程3389> get https://www.python.org <进程3390> get https://www.openstack.org <进程3388> get https://help.github.com/ <进程3387> parse https://www.baidu.com <进程3389> get http://www.sina.com.cn/ <进程3387> parse https://www.python.org <进程3387> parse https://help.github.com/ <进程3387> parse http://www.sina.com.cn/ <进程3387> parse https://www.openstack.org [{'url': 'https://www.baidu.com', 'text': '<!DOCTYPE html> ...',...}] '''
from multiprocessing import Pool import time,random import requests import re def get_page(url,pattern): response=requests.get(url) if response.status_code == 200: return (response.text,pattern) def parse_page(info): page_content,pattern=info res=re.findall(pattern,page_content) for item in res: dic={ 'index':item[0], 'title':item[1], 'actor':item[2].strip()[3:], 'time':item[3][5:], 'score':item[4]+item[5] } print(dic) if __name__ == '__main__': pattern1=re.compile(r'<dd>.*?board-index.*?>(d+)<.*?title="(.*?)".*?star.*?>(.*?)<.*?releasetime.*?>(.*?)<.*?integer.*?>(.*?)<.*?fraction.*?>(.*?)<',re.S) url_dic={ 'http://maoyan.com/board/7':pattern1, } p=Pool() res_l=[] for url,pattern in url_dic.items(): res=p.apply_async(get_page,args=(url,pattern),callback=parse_page) res_l.append(res) for i in res_l: i.get() # res=requests.get('http://maoyan.com/board/7') # print(re.findall(pattern,res.text))
如果在主进程中等待进程池中所有任务都执行完毕后，再统一处理结果，则无需回调函数
from multiprocessing import Pool import time,random,os def work(n): time.sleep(1) return n**2 if __name__ == '__main__': p=Pool() res_l=[] for i in range(10): res=p.apply_async(work,args=(i,)) res_l.append(res) p.close() p.join() #等待进程池中所有进程执行完毕 nums=[] for res in res_l: nums.append(res.get()) #拿到所有结果 print(nums) #主进程拿到所有的处理结果,可以在主进程中进行统一进行处理
进程池的其他实现方式：https://docs.python.org/dev/library/concurrent.futures.html
小结
需要注意一下
不能无限的开进程，不能无限的开线程
最常用的就是开进程池，开线程池。其中回调函数非常重要
回调函数其实可以作为一种编程思想，谁好了谁就去掉
只要你用并发，就会有锁的问题，但是你不能一直去自己加锁吧
那么我们就用QUEUE，这样还解决了自动加锁的问题
由Queue延伸出的一个点也非常重要的概念。以后写程序也会用到
这个思想。就是生产者与消费者问题
补充 ：必须掌握的八个DOS网络命令
一，ping 它是用来检查网络是否通畅或者网络连接速度的命令。作为一个生活在网络上的管理员或者黑客来说，ping命令是第一个必须掌握的DOS命令，它所利用的原理是这样的：网络上的机器都有唯一确定的IP地址，我们给目标IP地址发送一个数据包，对方就要返回一个同样大小的数据包，根据返回的数据包我们可以确定目标主机的存在，可以初步判断目标主机的操作系统等。下面就来看看它的一些常用的操作。先看看帮助吧，在DOS窗口中键入：ping /? 回车，。所示的帮助画面。在此，我们只掌握一些基本的很有用的参数就可以了（下同）。 -t 表示将不间断向目标IP发送数据包，直到我们强迫其停止。试想，如果你使用100M的宽带接入，而目标IP是56K的小猫，那么要不了多久，目标IP就因为承受不了这么多的数据而掉线，呵呵，一次攻击就这么简单的实现了。 -l 定义发送数据包的大小，默认为32字节，我们利用它可以最大定义到65500字节。结合上面介绍的-t参数一起使用，会有更好的效果哦。 -n 定义向目标IP发送数据包的次数，默认为3次。如果网络速度比较慢，3次对我们来说也浪费了不少时间，因为现在我们的目的仅仅是判断目标IP是否存在，那么就定义为一次吧。 说明一下，如果-t 参数和 -n参数一起使用，ping命令就以放在后面的参数为标准，比如"ping IP -t -n 3"，虽然使用了-t参数，但并不是一直ping下去，而是只ping 3次。另外，ping命令不一定非得ping IP，也可以直接ping主机域名，这样就可以得到主机的IP。 下面我们举个例子来说明一下具体用法。 这里time=2表示从发出数据包到接受到返回数据包所用的时间是2秒，从这里可以判断网络连接速度的大小 。从TTL的返回值可以初步判断被ping主机的操作系统，之所以说"初步判断"是因为这个值是可以修改的。这里TTL=32表示操作系统可能是win98。 （小知识：如果TTL=128，则表示目标主机可能是Win2000；如果TTL=250，则目标主机可能是Unix） 至于利用ping命令可以快速查找局域网故障，可以快速搜索最快的QQ服务器，可以对别人进行ping攻击……这些就靠大家自己发挥了。 （这个是比较早认识的命令了，因为对那些怎么侵入啊，就是黑客的东西特好奇，ping就听的早了。这次算是比较深的认识吧） 二，nbtstat 该命令使用TCP/IP上的NetBIOS显示协议统计和当前TCP/IP连接，使用这个命令你可以得到远程主机的NETBIOS信息，比如用户名、所属的工作组、网卡的MAC地址等。在此我们就有必要了解几个基本的参数。 -a 使用这个参数，只要你知道了远程主机的机器名称，就可以得到它的NETBIOS信息（下同）。 -A 这个参数也可以得到远程主机的NETBIOS信息，但需要你知道它的IP。 -n 列出本地机器的NETBIOS信息。 当得到了对方的IP或者机器名的时候，就可以使用nbtstat命令来进一步得到对方的信息了，这又增加了我们入侵的保险系数。 三，netstat 这是一个用来查看网络状态的命令，操作简便功能强大。 -a 查看本地机器的所有开放端口，可以有效发现和预防木马，可以知道机器所开的服务等信息，如图4。 这里可以看出本地机器开放有FTP服务、Telnet服务、邮件服务、WEB服务等。用法：netstat -a IP。 -r 列出当前的路由信息，告诉我们本地机器的网关、子网掩码等信息。用法：netstat -r IP。 四，tracert 跟踪路由信息，使用此命令可以查出数据从本地机器传输到目标主机所经过的所有途径，这对我们了解网络布局和结构很有帮助。 这里说明数据从本地机器传输到192.168.0.1的机器上，中间没有经过任何中转，说明这两台机器是在同一段局域网内。用法：tracert IP。 五，net 这个命令是网络命令中最重要的一个，必须透彻掌握它的每一个子命令的用法，因为它的功能实在是太强大了，这简直就是微软为我们提供的最好的入侵工具。首先让我们来看一看它都有那些子命令，键入net / 在这里，我们重点掌握几个入侵常用的子命令。 net view 使用此命令查看远程主机的所以共享资源。命令格式为net view \IP。 net use 把远程主机的某个共享资源影射为本地盘符，图形界面方便使用，呵呵。命令格式为net use x:\IPsharename。上面一个表示把192.168.0.5IP的共享名为magic的目录影射为本地的Z盘。下面表示和192.168.0.7建立IPC连接（net use \IPIPC "password" /user:"name"）， 建立了IPC连接后，呵呵，就可以上传文件了：copy nc.exe \192.168.0.7admin，表示把本地目录下的nc.exe传到远程主机，结合后面要介绍到的其他DOS命令就可以实现入侵了。 net start 使用它来启动远程主机上的服务。当你和远程主机建立连接后，如果发现它的什么服务没有启动，而你又想利用此服务怎么办？就使用这个命令来启动吧。用法：net start servername，如图9，成功启动了telnet服务。 net stop 入侵后发现远程主机的某个服务碍手碍脚，怎么办？利用这个命令停掉就ok了，用法和net start同。 net user 查看和帐户有关的情况，包括新建帐户、删除帐户、查看特定帐户、激活帐户、帐户禁用等。这对我们入侵是很有利的，最重要的，它为我们克隆帐户提供了前提。键入不带参数的net user，可以查看所有用户，包括已经禁用的。下面分别讲解。 1，net user abcd 1234 /add，新建一个用户名为abcd，密码为1234的帐户，默认为user组成员。 2，net user abcd /del，将用户名为abcd的用户删除。 3，net user abcd /active:no，将用户名为abcd的用户禁用。 4，net user abcd /active:yes，激活用户名为abcd的用户。 5，net user abcd，查看用户名为abcd的用户的情况 net localgroup 查看所有和用户组有关的信息和进行相关操作。键入不带参数的net localgroup即列出当前所有的用户组。在入侵过程中，我们一般利用它来把某个帐户提升为administrator组帐户，这样我们利用这个帐户就可以控制整个远程主机了。用法：net localgroup groupname username /add。 现在我们把刚才新建的用户abcd加到administrator组里去了，这时候abcd用户已经是超级管理员了，呵呵，你可以再使用net user abcd来查看他的状态，和图10进行比较就可以看出来。但这样太明显了，网管一看用户情况就能漏出破绽，所以这种方法只能对付菜鸟网管，但我们还得知道。现在的手段都是利用其他工具和手段克隆一个让网管看不出来的超级管理员，这是后话。有兴趣的朋友可以参照《黑客防线》第30期上的《由浅入深解析隆帐户》一文。 net time 这个命令可以查看远程主机当前的时间。如果你的目标只是进入到远程主机里面，那么也许就用不到这个命令了。但简单的入侵成功了，难道只是看看吗？我们需要进一步渗透。这就连远程主机当前的时间都需要知道，因为利用时间和其他手段（后面会讲到）可以实现某个命令和程序的定时启动，为我们进一步入侵打好基础。用法：net time \IP。 六，at 这个命令的作用是安排在特定日期或时间执行某个特定的命令和程序（知道net time的重要了吧？）。当我们知道了远程主机的当前时间，就可以利用此命令让其在以后的某个时间（比如2分钟后）执行某个程序和命令。用法：at time command \computer。 表示在6点55分时，让名称为a-01的计算机开启telnet服务（这里net start telnet即为开启telnet服务的命令）。 七，ftp 大家对这个命令应该比较熟悉了吧？网络上开放的ftp的主机很多，其中很大一部分是匿名的，也就是说任何人都可以登陆上去。现在如果你扫到了一台开放ftp服务的主机（一般都是开了21端口的机器），如果你还不会使用ftp的命令怎么办？下面就给出基本的ftp命令使用方法。 首先在命令行键入ftp回车，出现ftp的提示符，这时候可以键入"help"来查看帮助（任何DOS命令都可以使用此方法查看其帮助)。 大家可能看到了，这么多命令该怎么用？其实也用不到那么多，掌握几个基本的就够了。 首先是登陆过程，这就要用到open了，直接在ftp的提示符下输入"open 主机IP ftp端口"回车即可，一般端口默认都是21，可以不写。接着就是输入合法的用户名和密码进行登陆了，这里以匿名ftp为例介绍。 用户名和密码都是ftp，密码是不显示的。当提示**** logged in时，就说明登陆成功。这里因为是匿名登陆，所以用户显示为Anonymous。 接下来就要介绍具体命令的使用方法了。 dir 跟DOS命令一样，用于查看服务器的文件，直接敲上dir回车，就可以看到此ftp服务器上的文件。 cd 进入某个文件夹。 get 下载文件到本地机器。 put 上传文件到远程服务器。这就要看远程ftp服务器是否给了你可写的权限了，如果可以，呵呵，该怎么 利用就不多说了，大家就自由发挥去吧。 delete 删除远程ftp服务器上的文件。这也必须保证你有可写的权限。 bye 退出当前连接。 quit 同上。 八，telnet 功能强大的远程登陆命令，几乎所有的入侵者都喜欢用它，屡试不爽。为什么？它操作简单，如同使用自己的机器一样，只要你熟悉DOS命令，在成 功以administrator身份连接了远程机器后，就可以用它来**想干的一切了。下面介绍一下使用方法，首先键入telnet回车，再键入help查看其帮助信息。 然后在提示符下键入open IP回车，这时就出现了登陆窗口，让你输入合法的用户名和密码，这里输入任何密码都是不显示的。 当输入用户名和密码都正确后就成功建立了telnet连接，这时候你就在远程主机上具有了和此用户一样的权限，利用DOS命令就可以实现你想干的事情了。这里我使用的超级管理员权限登陆的。