主要内容:
线程的一些其他方法
线程事件
线程队列
线程池
GIL锁
协程
Greenlet
Gevent
一. 线程(threading)的一些其他方法
from threading import Thread import threading import time def work(): time.sleep(1) print("子线程对象>>>", threading.current_thread()) # 子线程对象 print("子线程名称>>>", threading.current_thread().getName()) # 子线程名称 print("子线程ID>>>", threading.get_ident()) # 子线程ID if __name__ == '__main__': t = Thread(target=work) # 创建子线程 t.start() # 开启子线程 print("主线程对象>>>", threading.current_thread()) # 主线程对象 print("主线程名称>>>", threading.current_thread().getName()) # 主线程名称 print("主线程ID>>>", threading.current_thread().ident) # 主线程ID print("主线程ID>>>", threading.get_ident()) # 主线程ID time.sleep(1) # 阻塞住,此时主线程代码运行的同时子线程代码也在运行 print(threading.enumerate()) # 拿到所有正在运行的线程对象(包括主线程) print(threading.active_count()) # 拿到所有正在运行的线程对象的数量 print("主线程/主进程执行完毕")
二. 线程事件
同进程的一样. 线程的一个关键特性是每个线程都是独立运行且状态不可预测。如果程序中的其他线程需要通过判断某个线程的状态来确定自己下一步的操作,这时线程同步问题就会变得非常棘手。为了解决这些问题,我们需要使用threading库中的Event对象。 对象包含一个可由线程设置的信号标志,它允许线程等待某些事件的发生。在初始情况下,Event对象中的信号标志被设置为假。如果有线程等待一个Event对象, 而这个Event对象的标志为假,那么这个线程将会被一直阻塞直至该标志为真。一个线程如果将一个Event对象的信号标志设置为真,它将唤醒所有等待这个Event对象的线程。如果一个线程等待一个已经被设置为真的Event对象,那么它将忽略这个事件, 继续执行.
事件的基本方法:
event.isSet()：返回event的状态值； event.wait()：如果 event.isSet()==False将阻塞线程； event.set()： 设置event的状态值为True，所有阻塞池的线程激活进入就绪状态， 等待操作系统调度； event.clear()：恢复event的状态值为False。
举例说明:
有多个工作线程尝试链接MySQL，我们想要在链接前确保MySQL服务正常才让那些工作线程去连接MySQL服务器，如果连接不成功，都会去尝试重新连接。我们现在采用threading.Event机制来协调各个工作线程的连接操作.
MySQL简述:
mysql就是一个数据库，存数据用的东西，它就像一个文件夹，里面存着很多的excel表格，我们可以在表格里面写数据，存数据。但是如果我们要使用数据库，我们必须先要去连接它，你和他建立了连接关系，你才能操作它里面存放的数据。
模拟一个场景,开启两个线程:
线程一: 连接数据库,这个线程需要等待一个信号,告诉我们双方之间的网络是可以连通的.
线程二:检测与数据库之间的网络是否联通,并发送一个可联通或者不可联通的信号.
from threading import Thread,Event import threading import time,random def conn_mysql(): count=1 while not event.is_set(): if count > 3: raise TimeoutError('链接超时') #自己发起错误 print('<%s>第%s次尝试链接' % (threading.current_thread().getName(), count)) event.wait(0.5) # count+=1 print('<%s>链接成功' %threading.current_thread().getName()) def check_mysql(): print(' 33[45m[%s]正在检查mysql 33[0m' % threading.current_thread().getName()) t1 = random.randint(0,3) print('>>>>',t1) time.sleep(t1) event.set() if __name__ == '__main__': event=Event() check = Thread(target=check_mysql) conn1=Thread(target=conn_mysql) conn2=Thread(target=conn_mysql) check.start() conn1.start() conn2.start()
三. 线程队列
queue队列: 使用import queue , 用法与进程Queue一样.
queue.Queue(maxsize=0) -- 先进先出
import queue #不需要通过threading模块里面导入，直接import queue就可以了，这是python自带的,用法基本和我们进程multiprocess中的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.get_nowait() #没有数据就报错，可以通过try来搞 # 执行结果: (先进先出) # first # second # third
queue.LifoQueue(maxsize=0) -- last in first out 后进先出
import queue q=queue.LifoQueue() #队列，类似于栈，后进先出的顺序 q.put('first') q.put('second') q.put('third') # q.put_nowait() print(q.get()) print(q.get()) print(q.get()) # q.get_nowait() # 执行结果:(后进先出) # third # second # first
queue.PriorityQueue(maxsize=0) -- 存储数据时可以设置优先级队列
rt queue q = queue.PriorityQueue() # 创建栈 # put()方法放置一个元组进入栈,元组的第一个元素是优先级(通常是数字,也可以是非数字之间的比较),数字越小优先级越高 q.put((-10, "a")) q.put((-5, "a")) # 负数也可以 # q.put((20,"ws")) # 如果两个值的优先级一样,那么按照后面的值的acsii码顺序来排序,如果字符串第一个数元素相同,比较第二个元素的acsii码顺序 # q.put((20,"wd")) # q.put((20,{"a": 11})) # TypeError: unorderable types: dict() < dict() 不能是字典 # q.put((20,("w", 1))) # 优先级相同的两个数据,他们后面的值必须是相同的数据类型才能比较,可以是元祖,也是通过元素的ascii码顺序来排序 q.put((20, "b")) 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()) # 数字越小优先级越高,优先级高的优先出队
四. 线程池
早期的时候并没有线程池,现在Python提供了一个新的标准或者说内置的模块,这个模块里面提供了新的线程池和进程池.
模块介绍:
# concurrent.futures模块提供了高度封装的异步调用接口 # ThreadPoolExecutor: 线程池,提供异步调用 # ProcessPoolExecutor: 进程池,提供异步调用 # 基本方法: # submit(func, *args, **kwargs) -- 异步提交任务 # map(func, *iterables, timeout=None, chunksize=1) -- 取代for循环submit的操作 # shutdown(wait=True) -- 相当于进程池的pool.close()和pool.join()操作 需要注意的是: wait=True,等待池内所有任务执行完毕回收完资源后才继续 wait=False,立即返回,并不会等待池内的任务执行完毕 但不管wait参数为何值,整个程序都会等到所有任务执行完毕 submit和map必须在shutdown之前 # result(timeout=None) -- 取得结果,相当于get(),如果没有值可取会阻塞住 # add_done_callback(func) -- 回调函数
ThreadPoolExecutor的简单使用
import time from concurrent.futures import ThreadPoolExecutor def func(n): time.sleep(2) return n**2 if __name__ == '__main__': thread_pool = ThreadPoolExecutor(max_workers=4) # 创建线程池对象,默认一般开启的线程数量不超过CPU个数的5倍 t_list = [] for i in range(20): t = thread_pool.submit(func, i) # 提交执行函数,返回一个结果对象,i作为任务函数的参数.submit(func, *args, **kwargs)可以传递任意形式的参数 t_list.append(t) # thread_pool.shutdown() # shutdown()的作用相当于close()和join(),等待所有的线程执行完毕 # for tt in t_list: # print(">>>", tt.result()) # 也可以不用shutdown()方法,改换下面这种方式: for n, tt in enumerate(t_list): # enumerate()枚举 print(">>>", n, tt.result()) time.sleep(2)
补充: ProcessPoolExecutor的使用
只需要将这一行代码改为下面这一行就可以了，其他的代码都不用变 tpool = ThreadPoolExecutor(max_workers=5) #默认一般起线程的数据不超过CPU个数的5倍 # tpool = ProcessPoolExecutor(max_workers=5) 你就会发现为什么将线程池和进程池都放到这一个模块里面了，用法一样
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor import threading import time, random def task(n): print("{} is runing".format(threading.get_ident())) # 子线程ID号 time.sleep(random.randint(1, 3)) return n**2 if __name__ == '__main__': executor = ThreadPoolExecutor(max_workers=3) # 创建线程池对象,设置的线程数量为3 for i in range(11): future = executor.submit(task, i) s = executor.map(task, range(1, 5)) # map()取代了 for循环+submit() print([i for i in s])
import time from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor def func(n): time.sleep(2) return n*n def call_back(m): print("结果为:{}".format(m.result())) if __name__ == '__main__': tpool = ThreadPoolExecutor(max_workers=5) # 创建进程池对象 t_list = [] for i in range(5): t = tpool.submit(func, i).add_done_callback(call_back)
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor 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 parse_page(res): res=res.result() print('<进程%s> parse %s' %(os.getpid(),res['url'])) parse_res='url:<%s> size:[%s] ' %(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) # for url in urls: # p.apply_async(get_page,args=(url,),callback=pasrse_page) # p.close() # p.join() p=ProcessPoolExecutor(3) for url in urls: p.submit(get_page,url).add_done_callback(parse_page) #parse_page拿到的是一个future对象obj，需要用obj.result()拿到结果
五. GIL锁