进程
multiprocess
Process —— 进程 在python中创建一个进程的模块startdaemon 守护进程join 等待子进程执行结束锁 Lock
acquire release
锁是一个同步控制的工具
如果同一时刻有多个进程同时执行一段代码，
那么在内存中的数据是不会发生冲突的
但是，如果涉及到文件，数据库就会发生资源冲突的问题
我们就需要用锁来把这段代码锁起来
任意一个进程执行了acquire之后，
其他所有的进程都会在这里阻塞，等待一个release信号量 semaphore
锁 + 计数器
同一时间只能有指定个数的进程执行同一段代码事件 Event
set clear is_set   控制对象的状态
wait  根据状态不同执行效果也不同状态是True ---> pass状态是False --> 阻塞
一般wait是和set clear放在不同的进程中
set/clear负责控制状态
wait负责感知状态
我可以在一个进程中控制另外一个或多个进程的运行情况IPC通信
队列 Queue
管道 PIPE

from multiprocessing import Process,Queue
q = Queue()
print(q.empty())　　

from multiprocessing import Process,Queue
q = Queue()
print(q.full())　　

from multiprocessing import Process,Queue
q = Queue(10)   # 创建一个只能放10个value的队列
for i in range(10):q.put(i)    # 增加一个value
print(q.qsize())    # 返回队列中目前项目的正确数量
print(q.full())     # 如果q已满，返回为True
q.put(111)  # 再增加一个值
print(q.empty())　　

import time
from multiprocessing import Process,Queue
def wahaha(q):print(q.get())q.put(2)    # 增加数字2if __name__ == '__main__':q = Queue()p = Process(target=wahaha,args=[q,])p.start()q.put(1)    # 增加数字1time.sleep(0.2)print(q.get())　　

from multiprocessing import Process, Queuedef producer(q, name, food):for i in range(5):print('{}生产了{}{}'.format(name, food, i))if __name__ == '__main__':q = Queue()Process(target=producer, args=(q, '康师傅', '红烧牛肉')).start()Process(target=producer, args=(q, '郑师傅', '红烧鱼块')).start()　　

import time
import random
from multiprocessing import Process, Queuedef producer(q, name, food):for i in range(5):time.sleep(random.random())  # 模拟生产时间print('{}生产了{}{}'.format(name, food, i))q.put('{}{}'.format(food, i))  # 放入队列def consumer(q, name):for i in range(10):food = q.get()  # 获取队列time.sleep(random.random())  # 模拟吃的时间print('{}吃了{}'.format(name, food))if __name__ == '__main__':q = Queue()Process(target=producer, args=(q, '康师傅', '红烧牛肉')).start()Process(target=producer, args=(q, '郑师傅', '红烧鱼块')).start()Process(target=consumer, args=(q, 'xiao')).start()　　

import time
import random
from multiprocessing import Process, Queuedef producer(q, name, food):for i in range(5):time.sleep(random.random())  # 模拟生产时间print('{}生产了{}{}'.format(name, food, i))q.put('{}{}'.format(food, i))  # 放入队列def consumer(q, name):for i in range(5):  # 修改为5，因为有2个人food = q.get()  # 获取队列time.sleep(random.random())  # 模拟吃的时间print('{}吃了{}'.format(name, food))if __name__ == '__main__':q = Queue()Process(target=producer, args=(q, '康师傅', '红烧牛肉')).start()Process(target=producer, args=(q, '郑师傅', '红烧鱼块')).start()Process(target=consumer, args=(q, 'xiao')).start()Process(target=consumer, args=(q, 'lin')).start()　　

import time
import random
from multiprocessing import Process, Queuedef producer(q, name, food):for i in range(5):time.sleep(random.random())  # 模拟生产时间print('{}生产了{}{}'.format(name, food, i))q.put('{}{}'.format(food, i))  # 放入队列def consumer(q, name):while True:food = q.get()  # 获取队列if food == 'done': break  # 当获取的值为done时，结束循环time.sleep(random.random())  # 模拟吃的时间print('{}吃了{}'.format(name, food))if __name__ == '__main__':q = Queue()  # 创建队列对象，如果不提供maxsize，则队列数无限制p1 = Process(target=producer, args=(q, '康师傅', '红烧牛肉'))p2 = Process(target=producer, args=(q, '郑师傅', '红烧鱼块'))p1.start()  # 启动进程p2.start()Process(target=consumer, args=(q, 'xiao')).start()Process(target=consumer, args=(q, 'lin')).start()p1.join()  # 保证子进程结束后再向下执行p2.join()q.put('done')  # 向队列添加一个值doneq.put('done')　　

JoinableQueue的实例p除了与Queue对象相同的方法之外，还具有以下方法：q.task_done()
使用者使用此方法发出信号，表示q.get()返回的项目已经被处理。如果调用此方法的次数大于从队列中删除的项目数量，将引发ValueError异常。q.join()
生产者将使用此方法进行阻塞，直到队列中所有项目均被处理。阻塞将持续到为队列中的每个项目均调用q.task_done()方法为止。
下面的例子说明如何建立永远运行的进程，使用和处理队列上的项目。生产者将项目放入队列，并等待它们被处理。

import time
import random
from multiprocessing import Process, JoinableQueuedef producer(q, name, food):for i in range(5):time.sleep(random.random())print('{}生产了{}{}'.format(name, food, i))q.put('{}{}'.format(food, i))q.join()  # 等到所有的数据都被task_done才结束def consumer(q, name):while True:food = q.get()  # 获取队列time.sleep(random.random())  # 模拟吃的时间print('{}吃了{}'.format(name, food))q.task_done()  # 向q.join()发送一次信号，证明一个数据已经被取走了if __name__ == '__main__':q = JoinableQueue()  # 创建可连接的共享进程队列# 生产者们：即厨师们p1 = Process(target=producer, args=(q, '康师傅', '红烧牛肉'))p2 = Process(target=producer, args=(q, '郑师傅', '红烧鱼块'))p1.start()  # 启动进程p2.start()# 消费者们：即吃货们c1 = Process(target=consumer, args=(q, 'xiao'))c2 = Process(target=consumer, args=(q, 'lin'))c1.daemon = True  # 设置守护进程c2.daemon = Truec1.start()  # 启动进程c2.start()p1.join()  # 保证子进程结束后再向下执行p2.join()　　

producerput生产完全部的数据就没有其他工作了在生产数据方：允许执行q.joinjoin会发起一个阻塞，直到所有当前队列中的数据都被消费
consumerget 获取到数据处理数据q.task_done()   告诉q，刚刚从q获取的数据已经处理完了consumer每完成一个任务就会给q发送一个taskdone
producer在所有的数据都生产完之后会执行q.join()
producer会等待consumer消费完数据才结束
主进程中对producer进程进行join
主进程中的代码会等待producer执行完才结束
producer结束就意味着主进程代码的结束
consumer作为守护进程结束结束顺序：
consumer中queue中的所有数据被消费
producer join结束
主今晨过的代码结束
consumer结束
主进程结束　　

#创建管道的类：
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, Pipedef f(conn):conn.send("Hello The_Third_Wave")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()pipe初使用

from multiprocessing import Pipe
left,right = Pipe()
left.send('1234')
print(right.recv())　　

from multiprocessing import Process,Pipe
def f(parent_conn,child_conn):parent_conn.close()     # 不写close将不会引发EOFErrorwhile True:try:print(child_conn.recv())except EOFError:child_conn.close()break
if __name__ == '__main__':# 在进程之间创建一条管道，并返回元组（conn1，conn2），其中conn1和conn2是表示管道两端的Connection对象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.send('hello')parent_conn.send('hello')parent_conn.close()p.join()    # 等待子进程结束　　

from multiprocessing import Process,Pipedef consumer(p,name):produce, consume=pproduce.close()while True:try:baozi=consume.recv()print('%s 收到包子:%s' %(name,baozi))except EOFError:breakdef producer(seq,p):produce, consume=pconsume.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('主进程')

pipe实现生产者消费者模型

from multiprocessing import Process,Pipe,Lockdef consumer(p,name,lock):produce, consume=pproduce.close()while True:lock.acquire()baozi=consume.recv()lock.release()if baozi:print('%s 收到包子:%s' %(name,baozi))else:consume.close()breakdef producer(p,n):produce, consume=pconsume.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('主进程')

多个消费之之间的竞争问题带来的数据不安全问题

进程间数据是独立的，可以借助于队列或管道实现通信，二者都是基于消息传递的
虽然进程间数据独立，但可以通过Manager实现数据共享，事实上Manager的功能远不止于此A manager object returned by Manager() controls a server process which holds Python objects and allows other processes to manipulate them using proxies.A manager returned by Manager() will support types list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Value and Array.

Manager模块介绍

from multiprocessing import Manager, Processdef func(dic):print(dic)if __name__ == '__main__':m = Manager()  # 创建一个server进程dic = m.dict({'count': 100})  # 这是一个特殊的字典p = Process(target=func, args=(dic,))p.start()p.join()　　

from multiprocessing import Manager, Processdef func(dic):dic['count'] = dic['count'] - 1print(dic)if __name__ == '__main__':m = Manager()  # 创建一个server进程dic = m.dict({'count': 100})  # 这是一个特殊的字典p = Process(target=func, args=(dic,))p.start()p.join()　　

from multiprocessing import Manager, Processdef func(dic):dic['count'] = dic['count'] - 1  # 每次减1if __name__ == '__main__':m = Manager()  # 创建一个server进程dic = m.dict({'count': 100})  # 这是一个特殊的字典p_lst = []  # 定义一个空列表for i in range(100):  # 启动100个进程p = Process(target=func, args=(dic,))p_lst.append(p)  # 进程追加到列表中p.start()  # 启动进程for p in p_lst: p.join()  # 等待100个进程全部结束print(dic)  # 打印dic的值　　

from multiprocessing import Manager, Process, Lockdef func(dic, lock):lock.acquire()  # 取得锁dic['count'] = dic['count'] - 1  # 每次减1lock.release()  # 释放锁if __name__ == '__main__':m = Manager()  # 创建一个server进程lock = Lock()  # 创建锁dic = m.dict({'count': 100})  # 这是一个特殊的字典p_lst = []  # 定义一个空列表for i in range(100):  # 启动100个进程p = Process(target=func, args=(dic, lock))p_lst.append(p)  # 进程追加到列表中p.start()  # 启动进程for p in p_lst: p.join()  # 等待100个进程全部结束print(dic)  # 打印dic的值　　

from multiprocessing import Manager, Process, Lockdef func(dic, lock):with lock:  # 上下文管理：必须有一个开始动作和一个结束动作的时候dic['count'] = dic['count'] - 1  # 每次减1if __name__ == '__main__':m = Manager()  # 创建一个server进程lock = Lock()  # 创建锁dic = m.dict({'count': 100})  # 这是一个特殊的字典p_lst = []  # 定义一个空列表for i in range(100):  # 启动100个进程p = Process(target=func, args=(dic, lock))p_lst.append(p)  # 进程追加到列表中p.start()  # 启动进程for p in p_lst: p.join()  # 等待100个进程全部结束print(dic)  # 打印dic的值　　

Pool([numprocess  [,initializer [, initargs]]]):创建进程池

1 numprocess:要创建的进程数，如果省略，将默认使用cpu_count()的值
2 initializer：是每个工作进程启动时要执行的可调用对象，默认为None
3 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 Pooldef fc(i):time.sleep(0.5)print('func%s' % i)if __name__ == '__main__':p = Pool(5)p.apply(func=fc, args=(1,))　　

import time
from multiprocessing import Pooldef fc(i):time.sleep(0.5)print('func%s' % i)if __name__ == '__main__':p = Pool(5)for i in range(5):p.apply(func=fc, args=(1,))  # 同步调用# p.apply_async(func=fc,args=(1,))    # 异步调用　　

import time
import random
from multiprocessing import Pooldef fc(i):print('func%s' % i)time.sleep(random.randint(1, 3))return i ** 2if __name__ == '__main__':p = Pool(5)  # 创建拥有5个进程数量的进程池ret_l = []for i in range(5):# p.apply(func=fc,args=(1,))  # 同步调用ret = p.apply_async(func=fc, args=(i,))  # 异步调用ret_l.append(ret)for ret in ret_l: print(ret.get())  # 打印返回结果　　

import os,time
from multiprocessing import Pooldef work(n):print('%s run' %os.getpid())time.sleep(3)return n**2if __name__ == '__main__':p=Pool(3) #进程池中从无到有创建三个进程,以后一直是这三个进程在执行任务res_l=[]for i in range(10):res=p.apply(work,args=(i,)) # 同步调用，直到本次任务执行完毕拿到res，等待任务work执行的过程中可能有阻塞也可能没有阻塞# 但不管该任务是否存在阻塞，同步调用都会在原地等着print(res_l)

进程池的同步调用

import os
import time
import random
from multiprocessing import Pooldef work(n):print('%s run' %os.getpid())time.sleep(random.random())return n**2if __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用法：如果使用异步提交的任务，主进程需要使用jion，等待进程池内任务都处理完，然后可以用get收集结果# 否则，主进程结束，进程池可能还没来得及执行，也就跟着一起结束了
    p.close()p.join()for res in res_l:print(res.get()) #使用get来获取apply_aync的结果,如果是apply,则没有get方法,因为apply是同步执行,立刻获取结果,也根本无需get

进程池的异步调用

# Pool内的进程数默认是cpu核数，假设为4（查看方法os.cup_count()）
# 开启6个客户端，会发现2个客户端处于等待状态
# 在每个进程内查看pid，会发现pid使用为4个，即多个客户端供用4个进程
from socket import *
from multiprocessing import Pool
import osserver = 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: breakconn.send(msg.upper())except Exception:breakif __name__ == '__main__':p = Pool(4)while True:conn, *_ = server.accept()p.apply_async(talk, args=(conn,))# p.apply(talk,args=(conn,client_addr))   # 同步的话，则同一时间只有一个客户端能访问

server：进程池版socket并发聊天

from socket import *client = socket(AF_INET, SOCK_STREAM)
client.connect(('127.0.0.1', 8080))
while True:msg = input('>>>').strip()if not msg: continueclient.send(msg.encode('utf-8'))msg = client.recv(1024)print(msg.decode('utf-8'))

client

需要回调函数的场景：进程池中任何一个任务一旦处理完了，就立即告知主进程：我好了额，你可以处理我的结果了。主进程则调用一个函数去处理该结果，该函数即回调函数我们可以把耗时间（阻塞）的任务放到进程池中，然后指定回调函数（主进程负责执行），这样主进程在执行回调函数时就省去了I/O的过程，直接拿到的是任务的结果。

import os
from urllib.request import urlopen
from multiprocessing import Pooldef get_url(url):print('-->',url,os.getpid(),'get_url进程')ret = urlopen(url)  # 打开urlcontent = ret.read()  # 读取网页内容return contentdef call(url):  # 回调函数#分析print(url,os.getpid(),'回调函数')if __name__ == '__main__':print(os.getpid(),'主进程')  # 主进程idl = ['http://www.baidu.com','http://www.sina.com','http://www.sohu.com','http://www.sogou.com','http://www.qq.com','http://www.bilibili.com',]p = Pool(5)ret_l = []for url in l:ret = p.apply_async(func=get_url,args=[url,],callback=call)  # 异步ret_l.append(ret)  # 将进程追加到列表中for ret in ret_l:ret.get()  # 获取进程返回值　　

import os
from urllib.request import urlopen
from multiprocessing import Pooldef get_url(url):print('-->',url,os.getpid(),'get_url进程')ret = urlopen(url)  # 打开urlcontent = ret.read()  # 读取网页内容return urldef call(url):  # 回调函数#分析print(url,os.getpid(),'回调函数')if __name__ == '__main__':print(os.getpid(),'主进程')  # 主进程idl = ['http://www.baidu.com','http://www.sina.com','http://www.sohu.com','http://www.sogou.com','http://www.qq.com','http://www.bilibili.com',]p = Pool(5)ret_l = []for url in l:ret = p.apply_async(func=get_url,args=[url,],callback=call)  # 异步ret_l.append(ret)  # 将进程追加到列表中for ret in ret_l:ret.get()  # 获取进程返回值

from multiprocessing import Pool
import requests
import json
import osdef 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]\n' %(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>\r\n...',...}]
'''

使用多进程请求多个url来减少网络等待浪费的时间

import re
from urllib.request import urlopen
from multiprocessing import Pooldef get_page(url,pattern):response=urlopen(url).read().decode('utf-8')return pattern,responsedef parse_page(info):pattern,page_content=infores=re.findall(pattern,page_content)for item in res:dic={'index':item[0].strip(),'title':item[1].strip(),'actor':item[2].strip(),'time':item[3].strip(),}print(dic)
if __name__ == '__main__':regex = r'<dd>.*?<.*?class="board-index.*?>(\d+)</i>.*?title="(.*?)".*?class="movie-item-info".*?<p class="star">(.*?)</p>.*?<p class="releasetime">(.*?)</p>'pattern1=re.compile(regex,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()

爬虫实例

import time
import random
from multiprocessing import Process, Queuedef consumer(q, name):while True:food = q.get()if food == 'done': breaktime.sleep(random.random())print('{}吃了{}'.format(name, food))def producer(q, name, food):for i in range(10):time.sleep(random.random())print('{}生产了{}{}'.format(name, food, i))q.put('{}{}'.format(food, i))if __name__ == '__main__':q = Queue()p1 = Process(target=producer, args=(q, 'Egon', '泔水'))p2 = Process(target=producer, args=(q, 'Yuan', '骨头鱼刺'))p1.start()p2.start()Process(target=consumer, args=(q, 'alex')).start()Process(target=consumer, args=(q, 'wusir')).start()p1.join()p2.join()q.put('done')q.put('done')