主要内容:
- 1.线程的一些其他方法
- 2.线程 -- 事件
- 3.,线程对列
- 4.线程池
- 5.GIL锁
- 6.协程
1.线程的一些其他方法
#Thread实例对象方法 # isAlive(): 返回线程是否活动的 # getName():返回线程名 # setName():设置线程名 #threading 模块提供的一些方法: #threading.currentThread() #返回当前线程变量 #threading.enumerate():返回一个包含正在运行的线程list,正在运行指线程启动后,结束前,不包括启动前和终止后的线程 #threading.activeCount():返回正在运行的线程数量,与len(threading.enumerate())有相同的结果
from threading import Thread import threading import time def work (): time.sleep(3) print("子线程",threading.get_ident()) #子线程ID print(threading.current_thread().getName()) #子线程名称 Thread-1 if __name__ == '__main__': t = Thread(target=work) t.start() print(threading.currentThread()) #主线程对象 <_MainThread(MainThread, started 14464)> print(threading.current_thread().getName()) #主线程名称 MainThread print (threading.currentThread().ident) #主线程 ID print(threading.get_ident()) #主线程 ID time.sleep(3) print(threading.enumerate()) #[<_MainThread(MainThread, started 14464)>, <Thread(Thread-1, started 5900)>] # print(threading.active_count()) #2
2.线程 -- 事件
(1)事件的基本方法
#事件基本方法 #event.isset():返回event的状态值 #event.wait():如果 event.isset() == False将阻塞线程 #event.clear():回复event的状态值为False
3.线程队列
(1) 先进先出队列
import queue q = queue.Queue() q.put("first") q.put("second ") q.put("third") #q.put_nowait() #没有数据就会报错,可以通过try来解决 print(q.get()) print(q.get()) print(q.get()) #q.put_nowait() #没有数据就会报错,可以通过try来解决
(2)先进后出队列
import queue q = queue.LifoQueue() #队列,类似于栈 q.put("first") q.put("second ") q.put("third") print(q.get()) print(q.get()) print(q.get())
(3)优先级队列
import queue q = queue.PriorityQueue() q.put((-10,"a")) q.put((-5,"a")) q.put((20,"ws")) q.put((20,"wd")) # q.put((20,{"a",11})) #TypeError: unorderable types: dict() < dict() 不能是字典 # q.put((20,("w",1))) #优先级相同的两个数据，他们后面的值必须是相同的数据类型才能比较，可以是元祖， # 也是通过元素的ascii码顺序来排序 q.put((20,'b')) # 通过元素的ascii码顺序来排序 q.put((20,'c')) q.put((20,'a')) q.put((0,'b')) q.put((30,'c')) print(q.get()) print(q.get()) print(q.get()) print(q.get()) # print(q.get()) # print(q.get()) print(q.get()) print(q.get()) print(q.get()) print(q.get()) print(q.get()) '''结果 (-10, 'a') (-5, 'a') (0, 'b') (20, 'a') (20, 'b') (20, 'c') (20, 'wd') (20, 'ws') (30, 'c')'''
4.线程池
(1) Python标准模块--concurrent.futures
# concurrent.futures 模块提供了高度封装的异步调用接口 #ThreadPoolExecutor:线程池,提供异步调用 #ProcessPoolExceutor:进程池.提供异步的调用 # Both implement the same interface, which is defined by the abstract Executor class # 两者都实现由抽象执行器类定义的同一接口。 #2.基本方法 #submit(fn,*args,**kwargs) #异步提交任务 #map(func,*iterable,timeout=None,chunksize = 1) #取代for循环submit的操作 #shutdown(wait = True) #相当于进程池的pool.close() + pool.join()操作 #wait = True 等待池内所有的任务执行完毕,回收完资源后才继续 #wait = False 立即返回,并不会等待池内任务的执行完毕,但不管wait参数为何值,张哥程序都会等到所有任务执行完毕 #submit 和 map 必须在 shutdown 之前 #result(timout =None ) #取得结果 #add_done_callback(fn) #回调函数
(2)线程池中的map方法
import time from concurrent.futures import ThreadPoolExecutor from threading import current_thread def func(n): time.sleep(1) print(n,current_thread().ident) if __name__ == '__main__': t_p = ThreadPoolExecutor(max_workers = 4) map_res = t_p.map(func,range(10)) #异步执行的,map自带join功能 print(map_res) print([i for i in map_res])
(3)其他方法示例
import time from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor from threading import current_thread def func(n): time.sleep(1) #此处等待一秒是为了将取值时候的打印结果进行区分 return n**2 def func2(n): print(n) if __name__ == '__main__': t_p = ThreadPoolExecutor(max_workers=4) t_res_lst = [] for i in range(10): res_obj =t_p.submit(func,i) #提交执行函数,返回一个结果对象,i作为任务函数的参数 t_res_lst.append(res_obj) t_p.shutdown() #起到原来的close阻止新任务进来 + join的作用，等待所有的线程执行完毕 print("t_res_lst",t_res_lst) #<Future at 0x1fa10a43400 state=finished returned int> 加了shutdown后全部变为finished #<Future at 0x19d37f5c7f0 state=running>, <Future at 0x19d37f5c9b0 state=pending> 不加shutdo时 for e_res in t_res_lst: print(e_res.result()) #在不加shutdown时,主线程在循环列表时,也是每一个进行取值, # 但是由于可以有四个对象可以一起取值,(拿前四个为例,) # 也就是当列表循环到4个时此时会等待将近1s # 因为在执行函数有停留,但线程池可以四个同时执行,因此会有4个结果一起出来 #但是当加了shutdown时,此时线程会全部执行完毕,然后在列表里是,此时的对象已经全部执行完毕
(4)回调函数
import time from concurrent.futures import ThreadPoolExecutor from threading import current_thread def func1(n): time.sleep(1) print(n,current_thread().getName()) return n**2 def func2(n): print(">>>>>>",n) print(">>>>>>",n.result()) if __name__ == '__main__': t_p = ThreadPoolExecutor(max_workers=4) for i in range(3): t_p.submit(func1,i).add_done_callback(func2) print("主线程结束")
5.GIL锁
(1)GIL锁的概述
python最早期的时候对于多线程也加锁，但是python比较极端的（在当时电脑cpu确实只有1核）加了一个GIL全局解释锁，是解释器级别的，锁的是整个线程，而不是线程里面的某些数据操作，每次只能有一个线程使用cpu，也就说多线程用不了多核，但是他不是python语言的问题，是CPython解释器的特性，如果用Jpython解释器是没有这个问题的，Cpython是默认的，因为速度快，Jpython是java开发的，在Cpython里面就是没办法用多核，这是python的弊病，历史问题
(2)GIL锁原理图解
但是有了这个锁我们就不能并发了吗？当我们的程序是偏计算的，也就是cpu占用率很高的程序（cpu一直在计算），就不行了，但是如果你的程序是I/O型的（一般你的程序都是这个）（input、访问网址网络延迟、打开/关闭文件读写），在什么情况下用的到高并发呢（金融计算会用到，人工智能（阿尔法狗），但是一般的业务场景用不到，爬网页，多用户网站、聊天软件、处理文件），I/O型的操作很少占用CPU，那么多线程还是可以并发的，因为cpu只是快速的调度线程，而线程里面并没有什么计算，就像一堆的网络请求，我cpu非常快速的一个一个的将你的多线程调度出去，你的线程就去执行I/O操作了
(3) 多进程与多线程执行效率的对比
import time from multiprocessing import Process from threading import Thread def func(): time.sleep(2) print('xxxxxxxx') if __name__ == '__main__': p_s_t = time.time() p_list = [] for i in range(10): p = Process(target=func,) p_list.append(p) p.start() [pp.join() for pp in p_list] p_e_t = time.time() p_dif_t = p_e_t - p_s_t t_s_t = time.time() t_list = [] for i in range(10): t = Thread(target=func,) t_list.append(t) t.start() [tt.join() for tt in t_list] t_e_t = time.time() t_dif_t = t_e_t - t_s_t print('多进程执行的时间',p_dif_t) # 2.34332537651062 print('多线程执行的时间',t_dif_t) # 2.002843141555786
import time from multiprocessing import Process from threading import Thread def func(): num = 0 for i in range(1,100000000): num += i # def func(): # time.sleep(2) # print('xxxxxxxx') if __name__ == '__main__': p_s_t = time.time() p_list = [] for i in range(10): p = Process(target=func,) p_list.append(p) p.start() [pp.join() for pp in p_list] p_e_t = time.time() p_dif_t = p_e_t - p_s_t t_s_t = time.time() t_list = [] for i in range(10): t = Thread(target=func,) t_list.append(t) t.start() [tt.join() for tt in t_list] t_e_t = time.time() t_dif_t = t_e_t - t_s_t print('多进程执行的时间',p_dif_t) # 31.72713589668274 print('多线程执行的时间',t_dif_t) # 51.51686716079712
6.协程
(1)协程的引入
本节的主题是基于单线程来实现并发，即只用一个主线程（很明显可利用的cpu只有一个）情况下实现并发，为此我们需要先回顾下并发的本质：切换+保存状态
cpu正在运行一个任务，会在两种情况下切走去执行其他的任务（切换由操作系统强制控制），一种情况是该任务发生了阻塞，另外一种情况是该任务计算的时间过长或有一个优先级更高的程序替代了它
协程本质上就是一个线程，以前线程任务的切换是由操作系统控制的，遇到I/O自动切换，现在我们用协程的目的就是较少操作系统切换的开销（开关线程，创建寄存器、堆栈等，在他们之间进行切换等），在我们自己的程序里面来控制任务的切换。
(2)基于生成器(yield)实现并发
#协程:单线程下实现并发 #并发:伪并行,遇到IO就切换,单核下多个任务之间切换执行,给你的效果就是貌似你的几个程序在同时执行.提高效率 #任务切换 + 保存状态 #并行:多核cpu,真正的同时执行 #串行:一个任务执行完在执行另外一个任务 #多线程多进程下的任务切换+保存状态是操作系统 #串行 import time def func1(): time.sleep(1) print("func1") def func2(): time.sleep(2) print("func2") if __name__ == '__main__': func1() func2() #基于yieid并发执行,多任务之间来回切换,这就是个简单的写成体现,但是他能够减少I/O时间吗?,不能 import time def consumer(): while True: x = yield time.sleep(1) print("处理了数据:",x) def producer(): g = consumer() # print("asdffdhjfds") next(g) #找到consumer函数的位置 for i in range(3): g.send(i) print("发送了数据",i) start = time.time() #基于yield保存的状态,实现两个任务直接来回切换,即并发效果 #如果每个人物中都加上打印,那么明显的看到两个任务的打印是你一次我一次,即并发执行 producer() #我在当前线程中只执行了这个函数,但是通过这个函数里面的send切换了另一个任务 stop = time.time() print(stop - start)
(3)greenlet的应用
import time from greenlet import greenlet def eat(name): print('%s eat 1' %name) #2 time.sleep(3) g2.switch('taibai') #3 print('%s eat 2' %name) #6 g2.switch() #7 def play(name): print('%s play 1' %name) #4 time.sleep(3) g1.switch() #5 print('%s play 2' %name) #8 g1=greenlet(eat) g2=greenlet(play) g1.switch('taibai')#可以在第一次switch时传入参数，以后都不需要
(4)genvent的使用(-------------------)
Homework