9.11 进程池与线程池
池子使用来限制并发的任务数目，限制我们的计算机在一个自己可承受的范围内去并发地执行任务
池子内什么时候装进程：并发的任务属于计算密集型 池子内什么时候装线程：并发的任务属于IO密集型
进程池：
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import time,os,random
def task(x):
print('%s 接客' %os.getpid())
time.sleep(random.randint(2,5))
return x**2
if __name__ == '__main__': # ProcessPoolExecutor创建并开启指定数目的进程
p=ProcessPoolExecutor() # 默认开启的进程数是cpu的核数
for i in range(20):
p.submit(task,i) # 一下并行执行四个任务，等其中一个任务执行完后再执行下一个
线程池：
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import time,os,random
def task(x):
print('%s 接客' %x)
time.sleep(random.randint(2,5))
return x**2
if __name__ == '__main__': # ThreadPoolExecutor创建并开启指定数目的线程
p=ThreadPoolExecutor(4) # 默认开启的线程数是cpu的核数*5
for i in range(20):
p.submit(task,i) # 一下并发执行四个任务，等其中一个任务执行完后再并发执行下一个
9.112 基于多线程实现并发的套接字通信(使用线程池)
服务端：
from socket import *
from threading import Thread
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
tpool=ThreadPoolExecutor(3) #ThreadPoolExecutor创建并开启指定数目的线程
def communicate(conn,client_addr):
while True: # 通讯循环
try:
data = conn.recv(1024)
if not data: break
conn.send(data.upper())
except ConnectionResetError:
break
conn.close()
def server():
server=socket(AF_INET,SOCK_STREAM)
server.bind(('127.0.0.1',8080))
server.listen(5)
while True: # 链接循环
conn,client_addr=server.accept()
print(client_addr)
# t=Thread(target=communicate,args=(conn,client_addr))
# t.start()
tpool.submit(communicate,conn,client_addr)#一下并发执行3个任务，等其中一个任务执行完后再并发执行下一个
server.close()
if __name__ == '__main__':
server()
客户端：
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'))
data=client.recv(1024)
print(data.decode('utf-8'))
client.close()
9.12 同步异步阻塞非阻塞
阻塞与非阻塞指的是程序的两种运行状态:
阻塞：遇到 I/O 就发生阻塞，程序一旦遇到阻塞操作就会停在原地，并且立刻释放CPU资源
非阻塞（就绪态或运行态）：没有遇到 I/O 操作，或者通过某种手段让程序即便是遇到 I/O 操作也不会停在原地，执行其他操作，力求尽可能多的占有CPU
同步与异步指的是提交任务的两种方式：
同步调用：提交完任务后，就在原地等待，直到任务运行完毕后，拿到任务的返回值，才继续执行下一行代码
异步调用：提交完任务后，不在原地等待，直接执行下一行代码
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import time,os,random
#from multiprocessing import Pool
def task(x):
print('%s 接客' %x)
time.sleep(random.randint(1,3))
return x**2
if __name__ == '__main__':
# 异步调用
p=ThreadPoolExecutor(4) # 默认开启的线程数是cpu的核数*5
obj_l=[]
for i in range(10):
obj=p.submit(task,i)
obj_l.append(obj)
# p.close()
# p.join()
p.shutdown(wait=True)# shutdown指的是不能再往进程池内提交任务,wait=True指等待进程池或线程池内所有的任务都运行完毕
print(obj_l[3].result()) # 9 #最后拿结果
print('主')
# 同步调用
p=ThreadPoolExecutor(4) # 默认开启的线程数是cpu的核数*5
for i in range(10):
print(p.submit(task,i).result())
print('主')
9.121 异步调用+回调机制
问题：
1、任务的返回值不能得到及时的处理，必须等到所有任务都运行完毕才能统一进行处理
2、解析的过程是串行执行的,如果解析一次需要花费2s,解析9次则需要花费18s
基于进程池：
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import requests
import os
import time
import random
def get(url):
print('%s GET %s' %(os.getpid(),url))
response=requests.get(url)
time.sleep(random.randint(1,3))
if response.status_code == 200:
return response.text
def pasrse(obj): # 干解析的活
res=obj.result() # 回调拿结果
print('%s 解析结果为：%s' %(os.getpid(),len(res))) # 4108 解析结果为：2443
if __name__ == '__main__':
urls=[
'https://www.baidu.com',
'https://www.baidu.com',
'https://www.baidu.com',
'https://www.baidu.com',
'https://www.baidu.com',
'https://www.baidu.com',
'https://www.baidu.com',
'https://www.baidu.com',
'https://www.python.org',
]
pool=ProcessPoolExecutor(4)
for url in urls:
obj=pool.submit(get,url) #parse函数会在obj对应的任务执行完毕后自动执行，会把obj自动传给parse
obj.add_done_callback(pasrse) #四个进程并发爬取信息，主进程在执行解析操作
print('主进程',os.getpid()) # 主进程 4108
基于线程池：
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
from threading import current_thread
import requests
import os
import time
import random
def get(url):
print('%s GET %s' %(current_thread().name,url))
response=requests.get(url)
time.sleep(random.randint(1,3))
if response.status_code == 200:
return response.text
def pasrse(obj): # 干解析的活
res=obj.result()
print('%s 解析结果为：%s' %(current_thread().name,len(res)))#ThreadPoolExecutor-0_1 解析结果为：
#
if __name__ == '__main__': #ThreadPoolExecutor-0_3 解析结果为：2443
urls=[
'https://www.baidu.com',
'https://www.baidu.com',
'https://www.baidu.com',
'https://www.baidu.com',
'https://www.baidu.com',
'https://www.baidu.com',
'https://www.baidu.com',
'https://www.baidu.com',
'https://www.python.org',
]
pool=ThreadPoolExecutor(4)
for url in urls:
obj=pool.submit(get,url) #parse函数会在obj对应的任务执行完毕后自动执行，会把obj自动传给parse
obj.add_done_callback(pasrse) #四个线程并发爬取信息，空闲者执行解析操作
print('主线程',current_thread().name) #主线程 MainThread
9.13 线程queue
队列：先进先出 queue.Queue()
import queue
q=queue.Queue(3)
q.put(1)
q.put(2)
q.put(3)
# q.put(4) 阻塞
print(q.get()) #
print(q.get()) #
print(q.get()) #
堆栈：后进先出 queue.LifoQueue()
import queue
q=queue.LifoQueue(3)
q.put('a')
q.put('b')
q.put('c')
print(q.get()) #c
print(q.get()) #b
print(q.get()) #a
优先级队列:可以以小元组的形式往队列里存值，第一个元素代表优先级，数字越小优先级越高
PriorityQueue()
import queue
q=queue.PriorityQueue(3)
q.put((10,'user1'))
q.put((-3,'user2'))
q.put((-2,'user3'))
print(q.get()) #(-3, 'user2')
print(q.get()) #(-2, 'user3')
print(q.get()) #(10, 'user1')
