python D32 管道线程池
Posted 我是一名劍客
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了python D32 管道线程池相关的知识，希望对你有一定的参考价值。
管道:
Pipe
Conn1,conn2 = Pipe()
数据共享:
M = Manager()
Dic = m.dict({‘name’:sbalex})
数据安全的问题
加锁
进程池 *****
Map:异步提交任务,参数是可迭代对象,自带close + join
Apply :同步提交任务,直接可以收到返回值
Apply_async() 异步提交任务:res.get() 阻塞效果
Close join
回调函数:callback=
一、管道
进程间通信（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异常。因此，在生产者中关闭管道不会有任何效果，除非消费者也关闭了相同的管道端点。
引发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的时候直接结束接收或者说结束循环，就不用每次都关闭各个进程中的管道了。
通过结束信号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实现数据共享，事实上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,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)
总结一下，进程之间的通信：队列、管道、数据共享也算
下面要讲的信号量和事件也相当于锁，也是全局的，所有进程都能拿到这些锁的状态，进程之间这些锁啊信号量啊事件啊等等的通信，其实底层还是socekt，只不过是基于文件的socket通信，而不是跟上面的数据共享啊空间共享啊之类的机制，我们之前学的是基于网络的socket通信，还记得socket的两个家族吗，一个文件的一个网络的，所以将来如果说这些锁之类的报错，可能你看到的就是类似于socket的错误，简单知道一下就可以啦~~~
工作中常用的是锁，信号量和事件不常用，但是信号量和事件面试的时候会问到，你能知道就行啦~~~
二 进程池和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
详解：apply_async和apply
#一：使用进程池（异步调用,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)
# s = res.get() #如果直接用res这个结果对象调用get方法获取结果的话，这个程序就变成了同步，因为get方法直接就在这里等着你创建的进程的结果，第一个进程创建了，并且去执行了，那么get就会等着第一个进程的结果，没有结果就一直等着，那么主进程的for循环是无法继续的，所以你会发现变成了同步的效果
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)
进程池版的socket并发聊天代码示例：
server端：tcp_server.py
#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)) #同步的话，则同一时间只有一个客户端能访问
client端：tcp_client.py
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个人进行聊天，因为进程池中只有4个进程可供调用，那有同学会问，我们这么多人想同时聊天怎么办，又不让用多进程，进程池也不能开太多的进程，那咋整啊，后面我们会学到多线程，到时候大家就知道了，现在你们先这样记住就好啦
然后我们再提一个回调函数
需要回调函数的场景：进程池中任何一个任务一旦处理完了，就立即告知主进程：我好了额，你可以处理我的结果了。主进程则调用一个函数去处理该结果，该函数即回调函数，这是进程池特有的，普通进程没有这个机制，但是我们也可以通过进程通信来拿到返回值，进程池的这个回调也是进程通信的机制完成的。 我们可以把耗时间（阻塞）的任务放到进程池中，然后指定回调函数（主进程负责执行），这样主进程在执行回调函数时就省去了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。
爬虫相关的requests模块简单使用
import requests response = requests.get(\'http://www.baidu.com\') print(response) print(response.status_code) #200正常，404找不到网页，503等5开头的是人家网站内部错误 print(response.content.decode(\'utf-8\'))
使用多进程请求多个url来减少网络等待浪费的时间
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]\\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...\',...}]
\'\'\'
爬虫示例
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) #主进程拿到所有的处理结果,可以在主进程中进行统一进行处理
进程池和信号量的区别：
进程池是多个需要被执行的任务在进程池外面排队等待获取进程对象去执行自己，而信号量是一堆进程等待着去执行一段逻辑代码。
信号量不能控制创建多少个进程，但是可以控制同时多少个进程能够执行，但是进程池能控制你可以创建多少个进程。
举例：就像那些开大车拉煤的，信号量是什么呢，就好比我只有五个车道，你每次只能过5辆车，但是不影响你创建100辆车，但是进程池相当于什么呢？相当于你只有5辆车，每次5个车拉东西，拉完你再把车放回来，给别的人拉煤用。
其他语言里面有更高级的进程池，在设置的时候，可以将进程池中的进程动态的创建出来，当需求增大的时候，就会自动在进程池中添加进程，需求小的时候，自动减少进程，并且可以设置进程数量的上线，最多为多，python里面没有。
以上是关于python D32 管道线程池的主要内容，如果未能解决你的问题，请参考以下文章
Python--线程队列(queue)multiprocessing模块（进程对列Queue管道(pipe)进程池）协程
python 管道 事件 信号量 进程池(map/同步/异步)回调函数