山西手机版建站系统哪家好,中国网湖北官网,wordpress优化方法,做网站复杂吗Asyncio是一个异步编程的框架#xff0c;可以解决异步编程#xff0c;协程调度问题#xff0c;线程问题#xff0c;是整个异步IO的解决方案。在学习asyncio之前#xff0c;我们先来理清楚同步/异步的概念#xff1a;同步是指完成事务的逻辑#xff0c;先执行第一个事务可以解决异步编程协程调度问题线程问题是整个异步IO的解决方案。在学习asyncio之前我们先来理清楚同步/异步的概念同步是指完成事务的逻辑先执行第一个事务如果阻塞了会一直等待直到这个事务完成再执行第二个事务顺序执行 一直等待异步是和同步相对的异步是指在处理调用这个事务的之后不会等待这个事务的处理结果直接处理第二个事务去了通过状态、通知、回调来通知调用者处理结果。异步IO采用消息循环的模式重复“读取消息—处理消息”的过程也就是说异步IO模型”需要一个消息循环在消息循环中主线程不断地重复“读取消息-处理消息”这一过程。event_loop 事件循环程序开启一个无限的循环程序员会把一些函数注册到事件循环上。当满足事件发生的时候调用相应的协程函数。coroutine 协程协程对象指一个使用async关键字定义的函数它的调用不会立即执行函数而是会返回一个协程对象。协程对象需要注册到事件循环由事件循环调用。task 任务一个协程对象就是一个原生可以挂起的函数任务则是对协程进一步封装其中包含任务的各种状态。async/await 关键字 用于定义协程的关键字async定义一个协程await用于挂起阻塞的异步调用接口。(async和await这两个关键词是在python3.5开始正式提出定义asyncio是python解决异步io编程的一个完整框架。关于定义和原理请参考官方文档: 协程与任务, 如何理解await)其中协程编程离不开的三大要点事件循环回调epoll/select(IO多路复用)以下内容有删减的摘自 : Asyncio并发编程www.langzi.fun事件循环简单案例(访问一个网站)async def get_url_title(url):
# 使用关键词async定义一个协程print(开始访问网站:{}.format(url))await asyncio.sleep(2)# 这一步至关重要# asyncio.sleep(2) 功能:异步非阻塞等待2s作用是模拟访问网站消耗的时间# await 的作用类似 yield即这个时候把线程资源控制权交出去,监听这个描述符直到这个任务完成# await 后面只能接三种类型1. 协程:Python 协程属于 可等待 对象因此可以在其他协程中被等待:2. 任务:任务 被用来设置日程以便 并发 执行协程。(当一个协程通过 asyncio.create_task() 等函数被打包为一个 任务该协程将自动排入日程准备立即运行)3. Future 对象:Future 是一种特殊的 低层级 可等待对象表示一个异步操作的 最终结果。(当一个 Future 对象 被等待这意味着协程将保持等待直到该 Future 对象在其他地方操作完毕。)如果await time.sleep(2) 是会报错的print(网站访问成功)if __name__ __main__:start_time time.time()loop asyncio.get_event_loop()# 一行代码创造事件循环loop.run_until_complete(get_url_title(http://www.langzi.fun))# 这是一个阻塞的方法,可以理解成多线程中的join方法# 直到get_url_title(http://www.langzi.fun)完成后才会继续执行下面的代码end_time time.time()print(消耗时间:{}.format(end_time-start_time))返回结果开始访问网站:http://www.langzi.fun
网站访问成功
消耗时间:2.0018768310546875简单案例(访问多个网站)协程的优势是多任务协作单任务访问网站没法发挥出他的功能一次性访问多个网站或者一次性等待多个IO响应时间才能发挥它的优势。# -*- coding:utf-8 -*-
import asyncio
import timeasync def get_url_title(url):print(开始访问网站:{}.format(url))await asyncio.sleep(2)print(网站访问成功)if __name__ __main__:start_time time.time()loop asyncio.get_event_loop()# 创造一个事件循环tasks [get_url_title(http://www.langzi.fun)for i in range(10)]# 这个列表代表总任务量即执行10次get_url_title()函数loop.run_until_complete(asyncio.wait(tasks))# asyncio.wait后面接上非空可迭代对象,一般来说是功能函数列表# 功能是一次性提交多个任务等待完成# loop.run_until_complete(asyncio.gather(*tasks))# 和上面代码功能一致但是gather更加高级如果是列表就需要加上*# 这里会等到全部的任务执行完后才会执行后面的代码end_time time.time()print(消耗时间:{}.format(end_time-start_time))对一个网站发起10次请求返回结果开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
消耗时间:2.0015649795532227gather与wait的区别gather更擅长于将函数聚合在一起wait更擅长筛选运行状况即gather更加高级他可以将任务分组也可以取消任务import asyncioasync def get_url_title(url):print(开始访问网站:{}.format(url))await asyncio.sleep(2)print(网站访问成功)return successif __name__ __main__:loop asyncio.get_event_loop()# 使用wait方法# tasks [get_url_title(http://www.langzi.fun)for i in range(10)]# loop.run_until_complete(asyncio.wait(tasks))# 使用gather方法实现分组导入(方法1)group1 [get_url_title(http://www.langzi.fun)for i in range(3)]group2 [get_url_title(http://www.baidu.com)for i in range(5)]loop.run_until_complete(asyncio.gather(*group1,*group2))# 这种方法会把两个全部一次性导入# 使用gather方法实现分组导入(方法2)group1 [get_url_title(http://www.langzi.fun)for i in range(3)]group2 [get_url_title(http://www.baidu.com)for i in range(5)]group1 asyncio.gather(*group1)group2 asyncio.gather(*group2)#group2.cancel() 取消group2任务loop.run_until_complete(asyncio.gather(group1,group2))# 这种方法会先把group1导入然后导入group2返回结果开始访问网站:http://www.baidu.com
开始访问网站:http://www.baidu.com
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.baidu.com
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.baidu.com
开始访问网站:http://www.baidu.com
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.baidu.com
开始访问网站:http://www.baidu.com
开始访问网站:http://www.baidu.com
开始访问网站:http://www.baidu.com
开始访问网站:http://www.baidu.com
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功另外一种使用gather获取返回结果:import asyncioasync def get_url_title(url):print(开始访问网站:{}.format(url))await asyncio.sleep(2)print(网站访问成功)return successif __name__ __main__:loop asyncio.get_event_loop()# 使用gather方法传递任务获取结果group1 asyncio.ensure_future(get_url_title(http://www.langzi.fun))loop.run_until_complete(asyncio.gather(group1))# 如果不是列表就不需要加*print(group1.result())返回结果开始访问网站:http://www.langzi.fun
网站访问成功
success还有一些复杂的区别转移到python 异步协程中查看协程的调用和组合十分灵活尤其是对于结果的处理如何返回如何挂起需要逐渐积累经验和前瞻的设计。简单案例(获取返回值)# -*- coding:utf-8 -*-
import asyncio
import timeasync def get_url_title(url):print(开始访问网站:{}.format(url))await asyncio.sleep(2)print(网站访问成功)return successif __name__ __main__:start_time time.time()loop asyncio.get_event_loop()# 创建一个事件循环get_future loop.create_task(get_url_title(http://www.langzi.fun))#get_future asyncio.ensure_future(get_url_title(http://www.langzi.fun))# 这两行代码功能用法一模一样loop.run_until_complete(get_future)print(获取结果:{}.format(get_future.result()))# 获取结果end_time time.time()print(消耗时间:{}.format(end_time-start_time))返回结果开始访问网站:http://www.langzi.fun
网站访问成功
获取结果:success
消耗时间:2.0019724369049072如果是多个网址传入访问多个网址的返回值呢只需要把前面的知识点汇总一起即可使用if __name__ __main__:start_time time.time()loop asyncio.get_event_loop()# 创建一个事件循环tasks [loop.create_task(get_url_title(http://www.langzi.fun)) for i in range(10)]# 把所有要返回的函数加载到一个列表loop.run_until_complete(asyncio.wait(tasks))# 这里和上面用法一样print(获取结果:{}.format([x.result() for x in tasks]))# 因为结果都在一个列表在列表中取值即可end_time time.time()print(消耗时间:{}.format(end_time-start_time))返回结果开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
开始访问网站:http://www.langzi.fun
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
网站访问成功
获取结果:[success, success, success, success, success, success, success, success, success, success]
消耗时间:2.0016491413116455简单案例(回调函数)上面的例子是一个协程函数当这个协程函数的await xxx执行完毕后想要执行另一个函数后然后再返回这个协程函数的返回结果该这么做# -*- coding:utf-8 -*-
import asyncio
from functools import partial
# partial的功能是 固定函数参数返回一个新的函数。你可以这么理解from functools import partialdef go(x,y):return xyg partial(go,y2)print(g(1))
返回结果3g partial(go,x5,y2)print(g())
返回结果7
async def get_url_title(url):print(开始访问网站:{}.format(url))await asyncio.sleep(2)print(网站访问成功)# 当这个协程函数快要结束返回值的时候会调用下面的call_back函数# 等待call_back函数执行完毕后才返回这个协程函数的值return successdef call_back(future,url):# 注意这里必须要传递future参数因为这里的future即代表下面的get_future对象print(检测网址:{}状态正常.format(url))if __name__ __main__:loop asyncio.get_event_loop()# 创建一个事件循环get_future loop.create_task(get_url_title(http://www.langzi.fun))# 将一个任务注册到loop事件循环中get_future.add_done_callback(partial(call_back,url http://www.langzi.fun))# 这里是设置当上面的任务完成要返回结果的时候执行call_back函数# 注意call_back函数不能加上()也就意味着你只能依靠partial方法进行传递参数loop.run_until_complete(get_future)# 等待任务完成print(获取结果:{}.format(get_future.result()))# 获取结果返回结果开始访问网站:http://www.langzi.fun
网站访问成功
检测网址:http://www.langzi.fun状态正常
获取结果:success梳理协程函数必须要使用关键词async定义如果遇到了要等待的对象必须要使用await使用await后面的任务必须是可等待对象(三种主要类型: 协程, 任务 和 Future.)运行前必须要创建一个事件循环(loop asyncio.get_event_loop(),一行代码即可)然后把任务加载到该事件循环中即可如果需要获取协程函数的返回值需要使用loop.create_task()或asyncio.ensure_future()函数在最后使用.result()获取返回结果。如果想要把多个任务注册到loop中需要使用一个列表包含他们调用的时候使用asyncio.wait(list)取消协程任务存在多个任务协程想使用ctrl c退出协程使用例子讲解import asyncio
async def get_time_sleep(t):print(开始运行等待:{}s.format(t))await asyncio.sleep(t)print(运行结束)if __name__ __main__:loop asyncio.get_event_loop()# 创建一个事件循环task_1 get_time_sleep(1)task_2 get_time_sleep(2)task_3 get_time_sleep(3)tasks [task_1,task_2,task_3]# 三个协程任务加载到一个列表try:loop.run_until_complete(asyncio.wait(tasks))except KeyboardInterrupt:# 当检测到键盘输入 ctrl c的时候all_tasks asyncio.Task.all_tasks()# 获取注册到loop下的所有taskfor task in all_tasks:print(开始取消协程)task.cancel()# 取消该协程,如果取消成功则返回Trueloop.stop()# 停止循环loop.run_forever()# loop事件循环一直运行# 这两步必须要做finally:loop.close()# 关闭事件循环run_forever 会一直运行直到 stop 被调用但是你不能像下面这样调 stoploop.run_forever()
loop.stop()run_forever 不返回stop 永远也不会被调用。所以只能在协程中调 stopasync def do_some_work(loop, x):print(Waiting str(x))await asyncio.sleep(x)print(Done)loop.stop()这样并非没有问题假如有多个协程在 loop 里运行asyncio.ensure_future(do_some_work(loop, 1))
asyncio.ensure_future(do_some_work(loop, 3))loop.run_forever()第二个协程没结束loop 就停止了——被先结束的那个协程给停掉的。要解决这个问题可以用 gather 把多个协程合并成一个 future并添加回调然后在回调里再去停止 loop。async def do_some_work(loop, x):print(Waiting str(x))await asyncio.sleep(x)print(Done)def done_callback(loop, futu):loop.stop()loop asyncio.get_event_loop()futus asyncio.gather(do_some_work(loop, 1), do_some_work(loop, 3))
futus.add_done_callback(functools.partial(done_callback, loop))loop.run_forever()其实这基本上就是 run_until_complete 的实现了run_until_complete 在内部也是调用 run_forever。关于loop.close()简单来说loop 只要不关闭就还可以再运行。loop.run_until_complete(do_some_work(loop, 1))
loop.run_until_complete(do_some_work(loop, 3))
loop.close()但是如果关闭了就不能再运行了loop.run_until_complete(do_some_work(loop, 1))
loop.close()
loop.run_until_complete(do_some_work(loop, 3)) # 此处异常梳理通过gather()启动的协程任务是可以直接取消的并且还能获取取消是否成功可以通过 asyncio.Task.all_tasks()获取所有的协程任务如果使用run_forever()的话会一直运行只能通过loop.stop()停止协程相互嵌套import asyncio
async def sum_tion(x,y):print(开始执行传入参数相加:{} {}.format(x,y))await asyncio.sleep(1)# 模拟等待1Sreturn (xy)async def print_sum(x,y):result await sum_tion(x,y)print(result)if __name__ __main__:loop asyncio.get_event_loop()loop.run_until_complete(print_sum(1000,2000))loop.close()返回结果开始执行传入参数相加:1000 2000
3000执行流程run_until_complete运行会注册task协程print_sum并开启事件循环print_sum协程中嵌套了子协程此时print_sum协程暂停类似委托生成器转到子协程协程sum_tion中运行代码期间子协程需sleep1秒钟直接将结果反馈到event loop中即将控制权转回调用方而中间的print_sum暂停不操作1秒后调用方将控制权给到子协程调用方与子协程直接通信子协程执行接下来的代码直到再遇到wait此实例没有最后执行到return语句子协程向上级协程print_sum抛出异常StopIteration同时将return返回的值返回给上级协程print_sum中的result接收值print_sum继续执行暂时时后续的代码直到遇到return语句向 event loop 抛出StopIteration异常此时协程任务都已经执行完毕事件循环执行完成event loop the loop is stoppedclose事件循环。如果想要获取协程嵌套函数返回的值就必须使用回调import asyncio
async def sum_tion(x,y)-int:print(开始执行传入参数相加:{} {}.format(x,y))await asyncio.sleep(1)# 模拟等待1Sreturn (xy)async def print_sum(x,y):result await sum_tion(x,y)return resultdef callback(future):return future.result()if __name__ __main__:loop asyncio.get_event_loop()future loop.create_task(print_sum(100,200))# 如果想要获取嵌套协程返回的值就必须使用回调future.add_done_callback(callback)loop.run_until_complete(future)print(future.result())loop.close()返回结果开始执行传入参数相加:100 200
300定时启动任务asyncio提供定时启动协程任务通过call_soon,call_later,call_at实现他们的区别如下call_sooncall_soon是立即执行def callback(sleep_times):print(预计消耗时间 {} s.format(sleep_times))
def stoploop(loop):print(时间消耗完毕)loop.stop()if __name__ __main__:start_time time.time()loop asyncio.get_event_loop()# 创建一个事件循环loop.call_soon(callback,5)# 立即启动callback函数loop.call_soon(stoploop,loop)# 上面执行完毕后立即启动执行stoploop函数loop.run_forever()#要用这个run_forever运行因为没有传入协程print(总共耗时:{}.format(time.time()-start_time))返回结果预计消耗时间 5 s
时间消耗完毕
总共耗时:0.0010013580322265625call_latercall_later是设置一定时间启动执行def callback(sleep_times):print(预计消耗时间 {} s.format(sleep_times))
def stoploop(loop):print(时间消耗完毕)loop.stop()if __name__ __main__:start_time time.time()loop asyncio.get_event_loop()loop.call_later(1,callback,1.0)# 等待1秒后执行callback函数传入参数是1.0loop.call_later(5,stoploop,loop)# 等待5秒后执行stoploop函数传入参数是looploop.run_forever()print(总共耗时:{}.format(time.time()-start_time))返回结果预计消耗时间 1.0 s
时间消耗完毕
总共耗时:5.002613544464111call_atcall_at类似与call_later但是他指定的时间不再是传统意义上的时间而是loop的内部时钟时间效果和call_later一样 call_later内部其实调用了call_laterimport time
import asynciodef callback(loop):print(传入loop.time()时间为: {} s.format(loop.time()))
def stoploop(loop):print(时间消耗完毕)loop.stop()if __name__ __main__:start_time time.time()loop asyncio.get_event_loop()now loop.time()# loop内部的时钟时间loop.call_at(now1,callback,loop)# 等待loop内部时钟时间加上1s后执行callba函数传入参数为looploop.call_at(now3,callback,loop)# 等待loop内部时钟时间加上3s后执行callba函数传入参数为looploop.call_at(now5,stoploop,loop)# 等待loop内部时钟时间加上1s后执行stoploop函数传入参数为loop返回结果:传入loop.time()时间为: 3989.39 s
传入loop.time()时间为: 3991.39 s
时间消耗完毕
总共耗时:5.002060174942017call_soon_threadsafe 线程安全的call_sooncall_soon_threadsafe用法和call_soon一致。但在涉及多线程时 会使用它.梳理call_soon直接启动call_later自己定时启动call_at根据loop.time()内部的时间设置等待时间启动call_soon_threadsafe和call_soon方法一致是保证线程安全的他们都是比较底层的在正常使用时很少用到。结合线程池Asyncio是异步IO编程的解决方案异步IO是包括多线程多进程和协程的。所以asyncio是可以完成多线程多进程和协程的在开头说到协程是单线程的如果遇到阻塞的话会阻塞所有的代码任务所以是不能加入阻塞IO的但是比如requests库是阻塞的socket如果不设置setblocking(false)的话也是阻塞的这个时候可以放到一个线程中去做也是可以解决的即在协程中集成阻塞IO就加入多线程一起解决问题。用requests完成异步编程(使用线程池)from concurrent.futures import ThreadPoolExecutor
import requests
import asyncio
import time
import redef get_url_title(url):# 功能是获取网址的标题r requests.get(url)try:title re.search(title(.*?)/title,r.content.decode(),re.S|re.I).group(1)except Exception as e:title eprint(title)if __name__ __main__:start_time time.time()loop asyncio.get_event_loop()# 创建一个事件循环p ThreadPoolExecutor(5)# 创建一个线程池开启5个线程tasks [loop.run_in_executor(p,get_url_title,http://www.langzi.fun)for i in range(10)]# 这一步很重要使用loop.run_in_executor()函数:内部接受的是阻塞的线程池执行的函数传入的参数# 即对网站访问10次使用线程池访问loop.run_until_complete(asyncio.wait(tasks))# 等待所有的任务完成print(time.time()-start_time)返回结果Langzi - Never Setter 永不将就 - 致力于Python开发网络安全工具,分享Python底层与进阶知识漏洞扫描器开发与爬虫开发 Langzi - Never Setter 永不将就 - 致力于Python开发网络安全工具,分享Python底层与进阶知识漏洞扫描器开发与爬虫开发 Langzi - Never Setter 永不将就 - 致力于Python开发网络安全工具,分享Python底层与进阶知识漏洞扫描器开发与爬虫开发 Langzi - Never Setter 永不将就 - 致力于Python开发网络安全工具,分享Python底层与进阶知识漏洞扫描器开发与爬虫开发 Langzi - Never Setter 永不将就 - 致力于Python开发网络安全工具,分享Python底层与进阶知识漏洞扫描器开发与爬虫开发 Langzi - Never Setter 永不将就 - 致力于Python开发网络安全工具,分享Python底层与进阶知识漏洞扫描器开发与爬虫开发 Langzi - Never Setter 永不将就 - 致力于Python开发网络安全工具,分享Python底层与进阶知识漏洞扫描器开发与爬虫开发 Langzi - Never Setter 永不将就 - 致力于Python开发网络安全工具,分享Python底层与进阶知识漏洞扫描器开发与爬虫开发 Langzi - Never Setter 永不将就 - 致力于Python开发网络安全工具,分享Python底层与进阶知识漏洞扫描器开发与爬虫开发 Langzi - Never Setter 永不将就 - 致力于Python开发网络安全工具,分享Python底层与进阶知识漏洞扫描器开发与爬虫开发
5.589502334594727访问10次消耗时间为5.5s尝试将 p ThreadPoolExecutor(10)线程数量设置成10个线程消耗时间为4.6s改用从进程池p ProcessPoolExecutor(10)也是一样可以运行的不过10个进程消耗时间也是5.5s并且消耗更多的CPU资源。### 用socket完成异步编程(使用线程池)import asyncio
from concurrent.futures import ThreadPoolExecutor
import socket
from urllib.parse import urlparse
import time
import redef get_url(url):# 通过socket请求htmlurl urlparse(url)host url.netlocpath url.pathif path :path /# 建立socket连接client socket.socket(socket.AF_INET, socket.SOCK_STREAM)client.connect((host, 80))client.send(GET {} HTTP/1.1rnHost:{}rnConnection:closernrn.format(path, host).encode(utf8))data bwhile True:d client.recv(1024)if d:data delse:breakdata data.decode(utf8)html_data data.split(rnrn)[1]# 把请求头信息去掉 只要网页内容title re.search(title(.*?)/title,html_data,re.S|re.I).group(1)print(title)client.close()if __name__ __main__:start_time time.time()loop asyncio.get_event_loop()p ThreadPoolExecutor(3) # 线程池 放3个线程tasks [loop.run_in_executor(p,get_url,http://www.langzi.fun) for i in range(10)]loop.run_until_complete(asyncio.wait(tasks))print(last time:{}.format(time.time() - start_time))返回结果Langzi - Never Setter 永不将就 - 致力于Python开发网络安全工具,分享Python底层与进阶知识漏洞扫描器开发与爬虫开发 Langzi - Never Setter 永不将就 - 致力于Python开发网络安全工具,分享Python底层与进阶知识漏洞扫描器开发与爬虫开发 Langzi - Never Setter 永不将就 - 致力于Python开发网络安全工具,分享Python底层与进阶知识漏洞扫描器开发与爬虫开发 Langzi - Never Setter 永不将就 - 致力于Python开发网络安全工具,分享Python底层与进阶知识漏洞扫描器开发与爬虫开发 Langzi - Never Setter 永不将就 - 致力于Python开发网络安全工具,分享Python底层与进阶知识漏洞扫描器开发与爬虫开发 Langzi - Never Setter 永不将就 - 致力于Python开发网络安全工具,分享Python底层与进阶知识漏洞扫描器开发与爬虫开发 Langzi - Never Setter 永不将就 - 致力于Python开发网络安全工具,分享Python底层与进阶知识漏洞扫描器开发与爬虫开发 Langzi - Never Setter 永不将就 - 致力于Python开发网络安全工具,分享Python底层与进阶知识漏洞扫描器开发与爬虫开发 Langzi - Never Setter 永不将就 - 致力于Python开发网络安全工具,分享Python底层与进阶知识漏洞扫描器开发与爬虫开发 Langzi - Never Setter 永不将就 - 致力于Python开发网络安全工具,分享Python底层与进阶知识漏洞扫描器开发与爬虫开发
last time:5.132313966751099使用socket完成http请求(未使用线程池)import asyncio
from urllib.parse import urlparse
import timeasync def get_url(url):# 通过socket请求htmlurl urlparse(url)host url.netlocpath url.pathif path :path /# 建立socket连接reader, writer await asyncio.open_connection(host, 80) # 协程 与服务端建立连接writer.write(GET {} HTTP/1.1rnHost:{}rnConnection:closernrn.format(path, host).encode(utf8))all_lines []async for raw_line in reader: # __aiter__ __anext__魔法方法line raw_line.decode(utf8)all_lines.append(line)html n.join(all_lines)return htmlasync def main():tasks []tasks [asyncio.ensure_future(get_url(http://www.langzi.fun)) for i in range(10)]for task in asyncio.as_completed(tasks): # 完成一个 print一个result await taskprint(result)if __name__ __main__:start_time time.time()loop asyncio.get_event_loop()loop.run_until_complete(main())print(last time:{}.format(time.time() - start_time))asyncio协程和之前讲解的select事件循环原理是一样的梳理协程中遇到必须要使用阻塞任务的时候可以把阻塞代码放到线程池中运行线程池中的代码放到loop.run_in_executor()里面并且所有任务保存到列表最后通过loop.run_until_complate(asyncio.wait(任务列表))中运行asyncio能通过socket实现与服务端建立连接与多进程的结合既然异步协程和多进程对网络请求都有提升那么为什么不把二者结合起来呢在最新的 PyCon 2018 上来自 Facebook 的 John Reese 介绍了 asyncio 和 multiprocessing 各自的特点并开发了一个新的库叫做 aiomultiprocess这个库的安装方式是pip3 install aiomultiprocess需要 Python 3.6 及更高版本才可使用。使用这个库我们可以将上面的例子改写如下import asyncio
import aiohttp
import time
from aiomultiprocess import Poolstart time.time()async def get(url):session aiohttp.ClientSession()response await session.get(url)result await response.text()session.close()return resultasync def request():url http://127.0.0.1:5000urls [url for _ in range(100)]async with Pool() as pool:result await pool.map(get, urls)return resultcoroutine request()
task asyncio.ensure_future(coroutine)
loop asyncio.get_event_loop()
loop.run_until_complete(task)end time.time()
print(Cost time:, end - start)这样就会同时使用多进程和异步协程进行请求但在真实情况下我们在做爬取的时候遇到的情况千变万化一方面我们使用异步协程来防止阻塞另一方面我们使用 multiprocessing 来利用多核成倍加速节省时间其实还是非常可观的。同步与通信和多线程多进程任务一样协程也可以实现和需要进行同步与通信。简单例子(顺序启动多任务)协程是单线程的他的执行依赖于事件循环中最后的loop.run_until_complate()import asyncionum 0async def add():global numfor i in range(10):await asyncio.sleep(0.1)num i
async def desc():global numfor i in range(10):await asyncio.sleep(0.2)num - iif __name__ __main__:loop asyncio.get_event_loop()tasks [add(),desc()]loop.run_until_complete(asyncio.wait(tasks))# 这里执行顺序是先执行add函数然后执行desc函数# 所以最后的结果是0loop.close()print(num)返回结果0这里使用一个共有变量协程下不需要加锁。简单例子(Lock(锁))# -*- coding:utf-8 -*-
import asyncio
import functoolsdef unlock(lock):print(线程锁释放成功)lock.release()async def test(locker, lock):print(f{locker} 等待线程锁释放)# ---------------------------------# with await lock:# print(f{locker} 线程锁上锁)# ---------------------------------# 上面这两行代码等同于# ---------------------------------# await lock.acquire()# print(f{locker} 线程锁上锁)# lock.release()# ---------------------------------await lock.acquire()print(f{locker} 线程锁上锁)lock.release()print(f{locker} 线程锁释放)async def main(loop):lock asyncio.Lock()await lock.acquire()loop.call_later(0.5, functools.partial(unlock, lock))# call_later() 表达推迟一段时间的回调, 第一个参数是以秒为单位的延迟, 第二个参数是回调函数await asyncio.wait([test(任务 1 , lock), test(任务 2, lock)])if __name__ __main__:loop asyncio.get_event_loop()loop.run_until_complete(main(loop))loop.close()返回结果:任务 1 等待线程锁释放
任务 2 等待线程锁释放
线程锁释放成功
任务 1 线程锁上锁
任务 1 线程锁释放
任务 2 线程锁上锁
任务 2 线程锁释放简单例子(Semaphore(信号量))可以使用 Semaphore(信号量) 来控制并发访问的数量:import asyncio
from aiohttp import ClientSessionasync def fetch(sem,url):async with sem:# 最大访问数async with ClientSession() as session:async with session.get(url) as response:status response.statusres await response.text()print({}:{} .format(response.url, status))return resif __name__ __main__:loop asyncio.get_event_loop()url http://www.langzi.funsem asyncio.Semaphore(1000)# 设置最大并发数为1000tasks [loop.create_task(fetch(sem,url))for i in range(100)]# 对网站访问100次loop.run_until_complete(asyncio.wait(tasks))简单例子(Condition(条件))import asyncioasync def consumer(cond, name, second):# 消费者函数await asyncio.sleep(second)# 等待延迟with await cond:await cond.wait()print({}: 得到响应.format(name))async def producer(cond):await asyncio.sleep(2)for n in range(1, 3):with await cond:print(生产者 {} 号.format(n))cond.notify(nn) # 挨个通知单个消费者await asyncio.sleep(0.1)async def producer2(cond):await asyncio.sleep(2)with await cond:print(释放信号量通知所有消费者)cond.notify_all()# 一次性通知全部的消费者async def main(loop):condition asyncio.Condition()# 设置信号量task loop.create_task(producer(condition))# producer 和 producer2 是两个协程, 不能使用 call_later(), 需要用到 create_task() 把它们创建成一个 taskconsumers [consumer(condition, name, index) for index, name in enumerate((c1, c2))]await asyncio.wait(consumers)task.cancel()print(---分割线---)task loop.create_task(producer2(condition))consumers [consumer(condition, name, index) for index, name in enumerate((c1, c2))]await asyncio.wait(consumers)task.cancel()# 取消任务if __name__ __main__:loop asyncio.get_event_loop()loop.run_until_complete(main(loop))loop.close()返回结果生产者 1 号
c1: 得到响应
生产者 2 号
c2: 得到响应
---分割线---
释放信号量通知所有消费者
c1: 得到响应
c2: 得到响应简单例子(Event(事件))与 Lock(锁) 不同的是, 事件被触发的时候, 两个消费者不用获取锁, 就要尽快地执行下去了import asyncio
import functoolsdef set_event(event):print(开始设置事件)event.set()async def test(name, event):print({} 的事件未设置.format(name))await event.wait()print({} 的事件已设置.format(name))async def main(loop):event asyncio.Event()# 声明事件print(事件是否设置: {}.format(event.is_set()))loop.call_later(0.1, functools.partial(set_event, event))# 在0.1s后执行set_event()函数对事件进行设置await asyncio.wait([test(e1, event), test(e2, event)])print(最终事件状态: {}.format(event.is_set()))if __name__ __main__:loop asyncio.get_event_loop()loop.run_until_complete(main(loop))loop.close()返回结果事件是否设置: False
e1 的事件未设置
e2 的事件未设置
开始设置事件
e1 的事件已设置
e2 的事件已设置
最终事件状态: True简单例子(协程间通信)协程是单线程因此使用list、dict就可以实现通信而不会有线程安全问题当然可以使用asyncio.Queuefrom asyncio import Queue
queue Queue(maxsize3)
# queue的put和get需要用await举个例子import asyncio
from asyncio import Queue
import random
import string
q Queue(maxsize100)async def add():while 1:await q.put(random.choice(string.ascii_letters))async def desc():while 1:res await q.get()print(res)await asyncio.sleep(1)if __name__ __main__:loop asyncio.get_event_loop()loop.run_until_complete(asyncio.wait([add(),desc()]))loop.run_forever()返回结果:D
b
S
x
...加速asynciouvloop这个使用库可以有效的加速asyncio本库基于libuv也就是nodejs用的那个库。使用它也非常方便不过目前不支持windowsimport asyncio
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())没错就是2行代码就可以提速asyncio。tokio同样可以做异步资源循环import tokio
asyncio.set_event_loop_policy(tokio.EventLoopPolicy())参考:python异步编程之asyncio百万并发 - 三只松鼠 - 博客园www.cnblogs.comAsyncio并发编程www.langzi.fun
