python 之 并发编程(进程池与线程池同步异步阻塞非阻塞线程queue)
Posted mylu
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了python 之 并发编程(进程池与线程池同步异步阻塞非阻塞线程queue)相关的知识,希望对你有一定的参考价值。
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 解析结果为: #2443 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
队列:先进先出 queue.Queue()
import queue q=queue.Queue(3) ? q.put(1) q.put(2) q.put(3) # q.put(4) 阻塞 ? print(q.get()) #1 print(q.get()) #2 print(q.get()) #3
堆栈:后进先出 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‘)
以上是关于python 之 并发编程(进程池与线程池同步异步阻塞非阻塞线程queue)的主要内容,如果未能解决你的问题,请参考以下文章
python并发编程之进程池,线程池concurrent.futures
python全栈脱产第37天------进程池与线程池协程gevent模块单线程下实现并发的套接字通信
27 Apr 18 GIL 多进程多线程使用场景 线程互斥锁与GIL对比 基于多线程实现并发的套接字通信 进程池与线程池 同步异步阻塞非阻塞