当前位置: 开发笔记 > 编程语言 > 正文

Day33并发编程(三)线程池进程池,协程

作者：mobiledu2502889953 | 来源：互联网 | 2024-10-19 15:40

1.进程池和线程池开进程和开线程都需要消耗资源,只不过线程相比进程耗费的资源较小,但是计算机的硬件是有限制的,我们不能无限制的去开启进程或者线程.进程池和线程池能帮助我们在计算机承

1.进程池和线程池

　　开进程和开线程都需要消耗资源,只不过线程相比进程耗费的资源较小,但是计算机的硬件是有限制的,我们不能无限制的去开启进程或者线程.进程池和线程池能帮助我们在计算机承受的范围内最大限度的利用计算机

什么是池

　　在保证计算机硬件安全的情况下最大限度的利用计算机

　　池其实是降低了程序的运行效率,但是保证了计算机硬件的安全

怎么使用池

　　我们首先定义一个池子,在里边放入固定数量的进程或线程,只要有任务来了,就派一个进程或线程去处理任务,如果固定数量的进程或线程使用完了,那么接下来的任务就在外面等待,直到有进程或线程的任务执行完毕,外面的任务才可以拿到空闲的进程或线程执行,这样我们既可以同时使用多个进程或线程,又可以保证计算机硬件不会因为任务过多而导致死机等情况

注意:池子中的进程或线程的数量是固定的,可以自己定义数量,但是里边的进程或线程是不会动态改变的.例如工厂中有5个工人,一堆任务过来,他们只能同时执行5个,一个工人做完一个任务之后再做另一个任务,任务在变,但是工人是不会变的,即虽然任务在变,但是进程或线程只会创建一次,执行多少任务还是那几个,这样也节省了反复开闭进程线程的时间

#1 介绍
concurrent.futures模块提供了高度封装的异步调用接口
ThreadPoolExecutor&＃xff1a;线程池&＃xff0c;提供异步调用
ProcessPoolExecutor: 进程池&＃xff0c;提供异步调用
Both implement the same interface, which is defined by the abstract Executor class.#2 基本方法
#submit(fn, *args, **kwargs)
异步提交任务#map(func, *iterables, timeout&＃61;None, chunksize&＃61;1)
取代for循环submit的操作#shutdown(wait&＃61;True)
相当于进程池的pool.close()&＃43;pool.join()操作
wait&＃61;True&＃xff0c;等待池内所有任务执行完毕回收完资源后才继续
wait&＃61;False&＃xff0c;立即返回&＃xff0c;并不会等待池内的任务执行完毕
但不管wait参数为何值&＃xff0c;整个程序都会等到所有任务执行完毕
submit和map必须在shutdown之前#result(timeout&＃61;None)
取得结果#add_done_callback(fn)
回调函数# done()
判断某一个线程是否完成# cancle()
取消某个任务

1 from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
2 import time
3 import os
4
5 # pool &＃61; ThreadPoolExecutor(5) # 创建线程池对象,参数可以不传,默认参数是当前计算机cpu个数*5
6 pool &＃61; ProcessPoolExecutor(5) # 创建进程池对象,参数可以不传,默认参数是当前计算机cpu个数
7
8 def task(i):
9 print(&＃39;线程%s,进程号:%s&＃39;%(i,os.getpid()))
10 time.sleep(1) # 模拟IO
11 return i
12
13 &＃39;&＃39;&＃39;
14 提交任务的方式有两种
15 同步:提交任务之后,原地等待任务的返回值,期间不做任何事
16 异步:提交任务之后,不等待返回结果,继续执行下一行代码
17 &＃39;&＃39;&＃39;
18
19 def call_back(n):
20 print(&＃39;回调函数拿到了返回值%s&＃39;%n.result()) # 返回值&＃43;.result()取得结果
21 &＃39;&＃39;&＃39;
22 异步提交任务怎么拿到返回值:
23 异步回调机制:当异步提交的任务有结果后,会自动触发回调函数的执行
24 &＃39;&＃39;&＃39;
25
26 if __name__ &＃61;&＃61; &＃39;__main__&＃39;:
27 t_list &＃61; []
28 for i in range(20):
29 t &＃61; pool.submit(task,i)
30 t.add_done_callback(call_back) # 异步回调机制
31 print(os.cpu_count()) # 当前计算机是8核计算机

进程线程池模块使用

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutorimport os,time,random
def task(n):print(&＃39;%s is runing&＃39; %os.getpid())time.sleep(random.randint(1,3))return n**2if __name__ &＃61;&＃61; &＃39;__main__&＃39;:executor&＃61;ThreadPoolExecutor(max_workers&＃61;3)# for i in range(11):# future&＃61;executor.submit(task,i)
executor.map(task,range(1,12)) #map取代了for&＃43;submit

map的用法

map的用法

2.协程

　　进程是资源单位,线程是执行单位,开辟多进程和多线程都能实现并发,提高执行效率,最大化的利用计算机硬件的执行速度,但是无论是创建线程还是创建进程,都需要消耗一定的时间,随着对效率的追求,我们需要一种能够基于单线程就可以实现并发的需求,这样执行效率就能被再次提高.

　　并发的本质是:切换&＃43;保存状态

　　cpu的切换执行有两种情况,一是遇到阻塞(IO操作等),二是执行时间过长,自动切换到另一个任务,当我们使用线程时也会遇到上述两种情况,这是cpu就会切换到另一个线程执行,当前任务就会被搁置,不能最大化的利用cpu,这时候如果能在一个线程中开辟其他任务,当cpu在该线程遇到IO操作时,执行该线程中的其他任务,而不是执行其他线程,这就能最大化的提高代码的运行效率

什么是协程

　　协程就是能够在单线程下实现并发的程序

　　协程是由用户程序自己控制调度的,他不是python自带的,是我们程序员自己开发的,程序员通过代码来检测程序中的IO,一旦遇到IO自己通过代码来切换,让操作系统误认为这个程序没有任何IO,一直执行下去,如果程序中的任务没有都被执行,程序就只在运行态和就绪态之间切换,提高代码的运行效率

注意:当程序中的所有任务都处于IO阻塞状态,cpu还是会切换到其他程序运行,并不会死盯着这一个程序

协程的优点

　　1.协程的切换开销更小,是应用程序级别的,操作系统感知不到,因而更加轻量化

　　2.单线程内就能实现并发,能够最大限度的利用cpu

缺点

　　1.协程是在单线程下工作的,所以他和线程一样无法利用多核优势,我们可以多进程&＃43;多线程&＃43;多协程综合使用

　　2.协程是在但一个线程下使用的,因而一旦出现阻塞,会阻塞整个线程

特点:

　　1.可以在单线程内实现并发

　　2.修改共享数据不需要加锁

　　3.用户程序自己切换&＃43;保存状态

　　4.一个协程遇到IO操作会自动切换到其他协程

Gevent模块

协程的使用需要引入第三方的模块Gevent

g1&＃61;gevent.spawn(func,1,,2,3,x&＃61;4,y&＃61;5)创建一个协程对象g1&＃xff0c;spawn括号内第一个参数是函数名&＃xff0c;如eat&＃xff0c;后面可以有多个参数&＃xff0c;可以是位置实参或关键字实参&＃xff0c;都是传给函数eat的g2&＃61;gevent.spawn(func2)g1.join() #等待g1结束

g2.join() #等待g2结束#或者上述两步合作一步&＃xff1a;gevent.joinall([g1,g2])

g1.value#拿到func1的返回值

用法介绍

1 from gevent import monkey;monkey.patch_all()
2 # 需要将from gevent import monkey;monkey.patch_all()放到文件的开头
3 from gevent import spawn
4 import time
5
6 def func1():
7 print(&＃39;任务1开始&＃39;)
8 time.sleep(1)
9 print(&＃39;任务1结束&＃39;)
10
11 def func2():
12 print(&＃39;任务2开始&＃39;)
13 time.sleep(1)
14 print(&＃39;任务2结束&＃39;)
15
16 start &＃61; time.time()
17 g1 &＃61; spawn(func1)
18 g2 &＃61; spawn(func2)
19 g1.join()
20 g2.join() # 执行时间1.0054426193237305
21 # func1() # 函数形式正常执行
22 # func2() # 执行时间2.0061123371124268
23 print(time.time() - start)

协程执行时间的对比

TCP单线程利用协程实现并发

1 # 单线程利用协程实现并发服务端
2 from gevent import monkey;monkey.patch_all()
3 from gevent import spawn
4 import socket
5
6 server &＃61; socket.socket()
7 server.bind((&＃39;127.0.0.1&＃39;,8080))
8 server.listen()
9
10 def func1():
11 while True:
12 conn, addr &＃61; server.accept()
13 spawn(func2,conn)
14
15 def func2(conn):
16 while True:
17 try:
18 data &＃61; conn.recv(1024)
19 if len(data) &＃61;&＃61; 0:break
20 print(data.decode(&＃39;utf-8&＃39;))
21 conn.send(data.upper())
22 except ConnectionResetError as e:
23 print(e)
24 break
25 conn.close()
26
27 if __name__ &＃61;&＃61; &＃39;__main__&＃39;:
28 f1 &＃61; spawn(func1)
29 f1.join()

单线程利用协程实现并发服务端

1 # 单线程利用协程实现并发客户端
2 import socket
3 from threading import Thread,current_thread
4
5 client &＃61; socket.socket()
6 client.connect((&＃39;127.0.0.1&＃39;,8080))
7
8 def task(i):
9 while True:
10 data &＃61; &＃39;%s线程号%s&＃39;%(i,current_thread().name)
11 client.send(data.encode(&＃39;utf-8&＃39;))
12 res &＃61; client.recv(1024)
13 print(res.decode(&＃39;utf-8&＃39;))
14
15 i &＃61; 0
16 while True:
17 t &＃61; Thread(target&＃61;task,args&＃61;(i,))
18 t.start()
19 i &＃43;&＃61; 1