python之进程
一、什么是进程
进程（Process）是计算机中的程序关于某数据集合上的一次运行活动，是系统进行资源分配和调度的基本单位，是操作系统结构的基础。在早期面向进程设计的计算机结构中，进程是程序的基本执行实体；在当代面向线程设计的计算机结构中，进程是线程的容器。程序是指令、数据及其组织形式的描述，进程是程序的实体。我们自己在python文件中写了一些代码，这叫做程序，运行这个python文件的时候，这叫做进程。
狭义定义：进程是正在运行的程序的实例（an instance of a computer program that is being executed）。
SRE实战 互联网时代守护先锋，助力企业售后服务体系运筹帷幄！一键直达领取阿里云限量特价优惠。 广义定义：进程是一个具有一定独立功能的程序关于某个数据集合的一次运行活动。它是操作系统动态执行的基本单元，在传统的操作系统中，进程既是基本的分配单元，也是基本的执行单元 1，进程的概念 第一，进程是一个实体。每一个进程都有它自己的地址空间，一般情况下，包括文本区域（text region）（python的文件）、数据区域（data region）（python文件中定义的一些变量数据）和堆栈（stack region）。文本区域存储处理器执行的代码；数据区域存储变量和进程执行期间使用的动态分配的内存；堆栈区域存储着活动过程调用的指令和本地变量。 第二，进程是一个“执行中的程序”。程序是一个没有生命的实体，只有处理器赋予程序生命时（操作系统执行之），它才能成为一个活动的实体，我们称其为进程。进程是操作系统中最基本、重要的概念。 是多道程序系统出现后，为了刻画系统内部出现的动态情况，描述系统内部各道程序的活动规律引进的一个概念,所有多道程序设计操作系统都建立在进程的基础上。2，进程的特征 动态性：进程的实质是程序在多道程序系统中的一次执行过程，进程是动态产生，动态消亡的。 并发性：任何进程都可以同其他进程一起并发执行 独立性：进程是一个能独立运行的基本单位，同时也是系统分配资源和调度的独立单位； 异步性：由于进程间的相互制约，使进程具有执行的间断性，即进程按各自独立的、不可预知的速度向前推进 结构特征：进程由程序、数据和进程控制块三部分组成。 多个不同的进程可以包含相同的程序：一个程序在不同的数据集里就构成不同的进程，能得到不同的结果；但是执行过程中，程序不能发生改变。 注意：同一个程序执行两次，就会在操作系统中出现两个进程，所以我们可以同时运行一个软件，分别做不同的事情也不会混乱
二、并发与并行，同步与异步
并发：伪并行，就是几个进程共用一个cpu，几个进程之间是时间轮换的，而这个轮换时间很短，肉眼是看不到的，所以我们肉眼看到的是几个进程同时在进行，其他中间有停止的，只是停止时间短，这就叫并发
并行：指一个进程就用一个cpu，每个进程是没有停止的，就是一起运行的
同步：一个任务执行完后，另一个任务才能开始执行，就是有一个任务需要等待，这样两个任务之间的关系是同根生的，非常紧密的
异步：两个任务之间没有关联，你执行你的，我执行我的，互相不用等待
三、进程的创建方法
1，方法一
from multiprocessing import Process #引入进程模块， import os,time def fun1(): time.sleep(2) print('jjjj') def fun2(): time.sleep(3) print('ddada') print(os.getpid()) #获得子进程的进程id print(os.getppid()) #获得子进程的父进程，即主进程的id if __name__=="__main__": #记住，必须用 if main start_time=time.time() #开始时间 p=Process(target=fun2) #创建一个子进程 p.start() #开始子进程 fun1() print(os.getpid()) #获得主进程的进程id end_time=time.time() #结束时间 print(end_time-start_time) #打印总共用时
整段程序走完用时2秒多一点，但是如果我们按照以前的写法，先执行fun1（），再执行fun2（），至少得花5秒
2，方法二
from multiprocessing import Process #同样引入模块 import time class Mypro(Process): #创建一个类，继承Process def __init__(self,name): super().__init__() self.name=name def run(self): #这是必须的方法，然后要执行的程序写在方法里面就行 time.sleep(3) print(self.name) if __name__ == '__main__': #记住，这里必须写 if main p1=Mypro('huangchun') #用自己定义的类创建一个子进程 p1.start() #子进程开始 p1.join() #这是join方法，这样写就是join后面的主进程程序必须等子进程执行完后在执行，两种方法都可以用 print('我是主进程，哈哈哈哈哈哈。。。。。')
四、守护进程
把一个子进程设为守护进程之后，每当主进程执行完毕后，不管你子进程有没有执行完毕，都要跟着结束
from multiprocessing import Process import time def fun(): time.sleep(5) print('我是子进程') if __name__ == '__main__': p=Process(target=fun) p.daemon = True #这就是把子进程设为了守护进程，但是切记，这句代码必须写在start（）前面 p.start() print('我主进程，哈哈哈哈哈哈')
没有设置守护进程之前，代码要执行5秒才结束，等到子进程执行完才结束，但是设置为守护进程之后，代码瞬间执行完毕，只打印了主进程的内容，而子进程的print直接死了
五、同步锁（互斥锁）
对一段程序设置同步锁之后，不管你对这段程序开了几个进程，但只有一个进程能执行代码，而其他的进程需要等待前面的进程结束之后才能执行这段代码，就相当于把开的几个进程搞成同步了，互相约束着
这是模拟抢火车票的情景，在还没付钱确认之前，大家都看见了有一张票，但是只有一个人能抢到票
from multiprocessing import Process,Lock #引入Lockmok import time def fun1(i): #这是查看余票的方法 with open('piao.txt',mode='r',encoding='utf-8')as f1: num=f1.read() print('%s客户看到票还有%s张'%(i,num)) def fun(i,lo): #这是买票的方法 time.sleep(1) lo.acquire() #这是同步锁的开始， with open('piao.txt',mode='r',encoding='utf-8')as f1: num=f1.read() if num=='1': print('%s买到票了'%i) num=int(num)-1 elif num=='0': print('%s没票了'%i) with open('piao.txt', mode='w', encoding='utf-8')as f1: f1.write(str(num)) lo.release() 这是同步锁的结束 if __name__ == '__main__': lo=Lock() #这是创建一把锁 for i in range(10): #这是循环创建10个客户 fun1(i) #这是客户查看余票 p=Process(target=fun,args=(i,lo)) #创建子进程去抢票，然后把锁传给进程 p.start()
同步锁把抢票的程序锁起来了，所以，虽然有10个客户都在抢票，但实际上只有一人能进入抢票程序，其他的人要等待第一个进抢票程序的人执行完后才能又进一个人，反正每次就只允许一个人在使用抢票程序，其他的人继续排队
六、信号量
信号量就是对一段程序设定允许最多几个人使用，相当于一个饭店，只有四个位置，最多允许四个人吃饭，后面人要等待前面的人吃完才能进入饭店吃，前面走一个，就可以进一个
from multiprocessing import Process,Semaphore #引入Semaphore模块 import time def fun(s,i): s.acquire() print('%s客户在吃'%i) time.sleep(1) print('%s客户吃完了'%i) time.sleep(1) s.release() if __name__ == '__main__': s=Semaphore(4) #创建信号量对象 for i in range(10): #循环创建10个人去饭店吃饭 p=Process(target=fun,args=(s,i)) #创建饭店子进程，然后把信号量传给子进程 p.start()
七、事件
在创建事件之后默认值为False，我们可以把创建的事件对象传给子进程，然后在子进程中事件对象.set（），使得指变为Ture。然后我们在主进程中加上事件对象.wait()，在此处，只有当值为Ture时不会阻塞，值为False就会阻塞，
从而我们可以用此方法来影响主进程的执行。
from multiprocessing import Process,Event #引入Event模块 import time def func(e): print('子进程开始执行') time.sleep(3) print('子进程结束') e.set() #设置事件值为Ture，我们还可以通过e.clear()把值改为False if __name__ == '__main__': e=Event() #创建事件对象e，此时默认值为False p=Process(target=func,args=(e,)) #创建子进程，把e传给子进程 p.start() print('等待子进程结束后拿到值') e.wait() #当值为False会阻塞，当值为Ture是，不会阻塞，因为我们在子进程中最好把值改为Ture，所以wait（）以后的程序要等到子进程执行完才能执行 print('拿到值，执行主进程') time.sleep(1) print('主进程执行完毕')
八、队列，消费者生产者模型，joinablequeue
1，队列
队列就相当于一个容器，里面可以放数据，特点是先放进去先拿出来，即先进先出，
from multiprocessing import Queue #引入Queue模块 q=Queue(2) #创建一个队列对象，并给他设置容器大小，即能放几个数据 q.put(1) #put（）方法是往容器里放数据 q.put(2) q.put(3) #这是往容器里放第三个数据，由于只能放两个，所以此处会阻塞，不会报错，相当于死循环 q.put_nowait(4) #put_nowait（）这也是从容器里放数据的方法，但如果容器满了，不会阻塞，会直接报错 q.get() #get（）方法是从容器里拿数据 q.get() q.get() #这是从容器里拿第三个数据，但是容器的两个数据拿完了，没有数据了，此时也会阻塞，不会报错，死循环 q.get(False) #这也是从容器里拿数据的方法，当没数据时不会阻塞，直接报错 q.get_nowait() #这也是从容器里拿数据的方法，当没数据时不会阻塞，直接报错 q.full() #这是查看容器是否满了，如果满了，返回Ture，否则返回False q.empty() #这是查看容器是否为空，如果为空，返回Ture，否则返回False
2，消费者生产者模型
就是消费者和生产者之间不是直接联系的，而是通过中间的一个队列来进行联系的，生产者把生产的东西放进进队列里，消费者从队列里拿东西，其实消费者和生产者之间没有实质的联系
from multiprocessing import Process,Queue import time def sheng(q): #生产者 for i in range(10): time.sleep(1) q.put(i) print('包子%i生产完毕'%i) q.put('over') def xiao(q): #消费者 while 1: time.sleep(1.5) ss=q.get() if ss=='over': break print('包子%s被吃了'%ss) if __name__ == '__main__': q=Queue(10) #创建队列对象q p=Process(target=sheng,args=(q,)) #创建生产者子进程，把q传给他 p1=Process(target=xiao,args=(q,)) #创建消费者子进程，也把q传给他 p.start() p1.start(）
有个问题就是，生产者要把生产结束的消息放进队列里，让消费者过去结束消息，消费者才知道队列里没有数据了，这样消费者才能停止，如果生产者不放结束消息，当生产者结束时，即不往队列放数据，而消费者不知道生产者已经结束，
还一直往队列拿数据，当队列没数据时，消费者一边就会阻塞。解决阻塞，有几个消费者，生产者就应该放几个结束消息，让每个消费者都知道，这样每个消费者才能结束。但是生产者又不知道有几个消费者，所以不知奥应该放几个结束数据，
这样就无法解决对消费者现象。此时，我们就就可以引入joinablequeue
3，joinablequeue
他其实就是一种队列，但她又比队列要多两种方法，task_done()和join()方法，正是有这两种方法就可以解决上面的问题
from multiprocessing import Process,JoinableQueue #引入joinableQueue模块 import time def sheng(q): #生产者 for i in range(10): time.sleep(1) q.put(i) print('包子%i生产完毕'%i) q.join() #当q收到的task_done数量等于放进q的数据数量时，生产者就结束了 def xiao(q,i): #消费者 while 1: time.sleep(1.5) ss=q.get() if ss=='over': break print('%s客户吃包子%s'%(i,ss)) q.task_done() #消费者每从q里取一个值，就向q返回一个task_done消息 if __name__ == '__main__': q=JoinableQueue() #创建joinablequeue对象q p=Process(target=sheng,args=(q,)) #创建生产者子进程，把q传给他 p.start() for i in range(3): #循环创建消费者子进程，把q传给他 p1=Process(target=xiao,args=(q,i)) p1.daemon=True #把创建的每个消费者子进程设为守护进程 p1.start() p.join() #主进程要等到生产者子进程结束后才结束
整个个过程就是：生产者生产了10个包子，3个消费者吃包子，没吃一个包子往q里发一个task_done,当q拥有10个task_done时，意味着10个包子吃完了，此时，生产者就结束了，接着主进程也也结束了，然后守护进程跟着结束了，即所有的消费者子进程结束，解决上面所遇到的问题
九、管道
管道是双向通的，在创建管道时有两个端口，分别定义为conn1，conn2，而这两个端口又可以给多个进程，多个进程就可以通过管道通信。比如主进程的conn1.send()网管道里放数据，主进程的conn2.recv()可以拿值，或者子进程的conn2.recv()也可以拿值，但值只能被拿一次，拿了之后只就没了，还有我们是不可以用子进程的conn1.recv拿值。
管道是双向通的，在创建管道时有两个端口，分别定义为conn1，conn2，而这两个端口又可以给多个进程，多个进程就可以通过管道通信。比如主进程的conn1.send()网管道里放数据，主进程的conn2.recv()可以拿值，或者子进程的conn2.recv()也可以拿值，
但值只能被拿一次，拿了之后只就没了，还有我们是不可以用子进程的conn1.recv拿值。 from multiprocessing import Process,Pipe #引入Pipe模块 import time def fun(conn1,conn2): #子进程的方法 conn1.close() #关闭子进程的端口conn1 time.sleep(2) # ww=conn1.recv() #用子进程的端口conn1接收数据 # print(ww) ss=conn2.recv() #用子进程的端口conn2接收数据 print(ss) if __name__ == '__main__': conn1,conn2=Pipe() #创建一个管道，并定义两个端口分别为conn1，conn2 p=Process(target=fun,args=(conn1,conn2)) #创建一个子进程，并把管道的两个端口传给他 p.start() # conn2.close() #因为管道是在主进程中创建，所以直接在主进程用端口，就相当于主进程的管道端口，现在是把主进程的conn2端口给关闭 conn1.send('eeeee') #用主进程的conn1端口往管道里传值 # conn2.send('ggggg') #用主进程的conn2端口往管道里传值 注意：从conn1端口往管道传的值，只能通过conn2拿值，不管是谁的conn2，都可以拿；同样从从conn2传的值，只能从conn1中拿值，不管是谁的conn1. 当所有的conn1都关闭时，不管用谁的conn2传值，还是拿值都会报错
十、数据共享
数据共享就是进程间通信的一种方式，但数据共享也会像利用文件进行进程间通信时，数据会混乱，所以也要利用同步锁来使得同一时间只有一个进程能使用
from multiprocessing import Process,Manager,Lock def fen(l2,lo): #子进程 with lo: #同步锁，虽创建了100个进程，但同一时间只有1个进程能对数据共享对象进行操作 l2[0] -=1 #把列表的第0位做减1操作 if __name__ == '__main__': m=Manager() #创建数据共享对象 l2=m.list([100]) #为对象设定一种数据类型，我现在是用的列表 lo=Lock() l1=[] for i in range(100): #循环创建100个子进程 p=Process(target=fen,args=(l2,lo)) #把数据共享的对象传给子进程 p.start() l1.append(p) [pp.join() for pp in l1] #这步是让主进程等待所有子进程执行完毕 print(l2[0])
十一、进程池
为什么要有进程池?进程池的概念。
在程序实际处理问题过程中，忙时会有成千上万的任务需要被执行，闲时可能只有零星任务。那么在成千上万个任务需要被执行的时候，我们就需要去创建成千上万个进程么？首先，创建进程需要消耗时间，销毁进程（空间，变量，文件信息等等的内容）也需要消耗时间。第二即便开启了成千上万的进程，操作系统也不能让他们同时执行，维护一个很大的进程列表的同时，调度的时候，还需要进行切换并且记录每个进程的执行节点，也就是记录上下文（各种变量等等乱七八糟的东西，虽然你看不到，但是操作系统都要做），这样反而会影响程序的效率。因此我们不能无限制的根据任务开启或者结束进程。就看我们上面的一些代码例子，你会发现有些程序是不是执行的时候比较慢才出结果，就是这个原因，那么我们要怎么做呢？
在这里，要给大家介绍一个进程池的概念，定义一个池子，在里面放上固定数量的进程，有需求来了，就拿一个池中的进程来处理任务，等到处理完毕，进程并不关闭，而是将进程再放回进程池中继续等待任务。如果有很多任务需要执行，池中的进程数量不够，任务就要等待之前的进程执行任务完毕归来，拿到空闲进程才能继续执行。也就是说，池中进程的数量是固定的，那么同一时间最多有固定数量的进程在运行。这样不会增加操作系统的调度难度，还节省了开闭进程的时间，也一定程度上能够实现并发效果
在进程池中，当主进程代码执行完，不管子进程有没有执行完都会结束，相当于进程池中的子进程都变为了守护进程，会跟随主进程的结束而结束
from multiprocessing import Process,Pool import time def fun(i): sum=0 for j in range(5): sum +=i if __name__ == '__main__': p=Pool(4) #创建进程池，设定进程池中最多允许4个进程在执行 starr_time=time.time() #这是没用进程池 l1=[] for i in range(500): p1=Process(target=fun,args=(i,)) l1.append(p1) p1.start() [pp.join() for pp in l1] end_time=time.time() print(‘没用’，end_time-starr_time) s_time=time.time() #从这里是开始用进程池处理任务 p.map(fun,range(500)) #用的是进程池的map（）方法，后面的一个参数必须是可迭代的对象，而且自带进程池close（）和join（）方法 e_time=time.time() print(‘用’，e_time-s_time) 最后打印结果是‘没用’是15秒左右，‘用’0.00099秒左右，差距是很大的，进程池的作用就凸显的很大，为什么会产生这样的差距，使用进程池只需创建最开始的4个进程，而没用进程池需要创建500个进程，这之间相差很大，这是其中一个原因
进程池同步方法：就是把进程搞成同步的，只有一个执行完了，下一个进程才能执行，主要运用apply（）方法
from multiprocessing import Process,Pool import time def fun(i): sum=0 time.sleep(1) for n in range(5): sum += i return sum if __name__ == '__main__': p=Pool(4) for i in range(10): res=p.apply(fun,args=(i,)) #调用进程池的apply（）方法，虽是10个进程，但是只能一个接着一个执行，此时返回的一个值 print(res)
进程池异步方法：运用apply_async()方法
from multiprocessing import Pool,Process import time def fun(i): sum=0 time.sleep(1) for n in range(5): sum +=i return sum if __name__ == '__main__': p=Pool(4) l1=[] for i in range(10): res=p.apply_async(fun,args=(i,)) #调用进程池的apply_async()方法，由于进程池只能允许有4个进程，所以，相当于同时有4个进程在进程池执行，但现在返回的不是值，而是一个对象 l1.append(res) #我们把对象放在一个列表，等所有程序结束再从列表拿值，如现在就用res.get()去拿值，就会形成阻塞，从而演变成进程一个接一个执行，变成同步执行的 p.close() p.join() for res in l1: print(res.get())
回调函数：callback,当一个函数需要子进程执行完后的值作为参数，我们就可以把这个函数搞成回调函数
from multiprocessing import Pool import time def fun1(a,b): return a+b def fun2(i): print('我是回调函数，%s'%i) if __name__ == '__main__': p=Pool(4) p.apply_async(fun1,args=(2,3),callback=fun2) #这样会把fun1的返回值作参数传入fun2中 time.sleep(2) 但一个问题是，若不在主进程中写入一个延迟2秒，当主进程代码走完，而子进程还没执行完，从而子进程会随着主进程结束而结束，从而回调函数也死亡
十二、多进程版socket和进程池版socket
多进程的socket服务器和进程池版可以实现几个客户端同时连接，但在服务端的子进程的方法中不能使用input（）方法
1，多进程版
服务端
from multiprocessing import Process import socket server=socket.socket() server.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1) ip_port=('127.0.0.1',8888) server.bind(ip_port) server.listen() def fun(conn): while 1: msg=conn.recv(1024) if msg.decode('utf-8')=='over': break print(msg.decode('utf-8')) conn.send(msg) conn.close() if __name__ == '__main__': while 1: conn,addr=server.accept() p=Process(target=fun,args=(conn,)) p.start()
客户顿，就和之前的一样
import socket
client=socket.socket()
ip_port=('127.0.0.1',8888)
client.connect(ip_port)
while 1:
msg=input('请输入：')
client.send(msg.encode('utf-8'))
if msg=='over':
break
msg1=client.recv(1024)
print(msg1.decode('utf-8'))
client.close()
2，进程池版
服务端
#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'))
十三，进程池版爬虫
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))
更多精彩
