Python: asyncio实战
Python 实战 asyncio
2023-09-11 14:16:16 时间
Server端设置3s才返回数据
单线程
import requests, random, string, datetime def download(url): print('start') response = requests.get(url) print('complete') lst = url.rsplit(sep = '/', maxsplit = 1) file = 'images/{}-{}'.format(random.choice(string.ascii_lowercase), lst[-1]) with open(file, mode = 'w+b') as f: f.write(response.content) if __name__ == '__main__': urls = ['http://localhost:555/fake.txt'] * 3 commence = datetime.datetime.now() for url in urls: download(url) closure = datetime.datetime.now() print(f'elapsed: {(closure - commence).total_seconds()}')
aiohttp & asyncio.create_task
import requests, random, string, datetime, asyncio, aiohttp async def fetch(session: aiohttp.ClientSession, url): print(f'fetch {url}') async with session.get(url, verify_ssl = False) as response: content = await response.content.read() file_name = 'images/{}-{}'.format(random.choice(string.ascii_lowercase), url.rsplit(sep = '/', maxsplit = 1)[-1]) with open(file = file_name, mode = 'w+b') as f: f.write(content) print(f'fetch done') return 55 async def main(): async with aiohttp.ClientSession() as session: urls = ['http://localhost:555/fake.txt'] * 3 tasks = [asyncio.create_task(fetch(session, url)) for url in urls] await asyncio.wait(tasks) res = [task.result() for task in tasks] print(res) if __name__ == '__main__': commence = datetime.datetime.now() asyncio.run(main()) closure = datetime.datetime.now() print(f'elapsed: {(closure - commence).total_seconds()}')
asyncio+concurrent.futures.ThreadPoolExecutor+concurrent.futures.ProcessPoolExecutor 解决不支持异步的函数
DeprecationWarning: 底层方法
import asyncio, concurrent.futures, requests, random, string async def download(url: str): print('commence', url) loop = asyncio.get_event_loop() future = loop.run_in_executor(None, requests.get, url) response = await future print('closure') file_name = 'images/{1}-{0}'.format(url.rsplit(sep = '/', maxsplit = 1)[-1], random.choice(string.ascii_lowercase)) with open(file_name, mode = 'wb') as f: f.write(response.content) if __name__ == '__main__': urls = ['http://localhost:555/fake.txt'] * 3 tasks = [download(url) for url in urls] loop = asyncio.get_event_loop() loop.run_until_complete(asyncio.wait(tasks))
NEW API
import asyncio, concurrent.futures, requests, random, string async def download(url: str): print('commence', url) loop = asyncio.get_event_loop() future = loop.run_in_executor(None, requests.get, url) response = await future print('closure') file_name = 'images/{1}-{0}'.format(url.rsplit(sep = '/', maxsplit = 1)[-1], random.choice(string.ascii_lowercase)) with open(file_name, mode = 'wb') as f: f.write(response.content) return 55 if __name__ == '__main__': urls = ['http://localhost:555/fake.txt'] * 3 tasks = [download(url) for url in urls] result = asyncio.run(asyncio.wait(tasks)) print(tasks) print(result)
相关文章
- 运行python程序时出现错误“SyntaxError: Non-ASCII character 'xe8' in file variable.py on line 1, but no encod”
- 5 分钟,教你用 Docker 部署一个 Python 应用
- Python数据可视化三部曲之 Pyecharts 从上手到上头
- Python __missing__ 魔法方法
- python 保存图片_用Python从视频中提取每一帧的图片
- 超全Python学习路线图+14张思维导图,让python初学者不走弯路
- 使用python生成随机数(random模块)
- 【python实战】拒绝黄牛!我要用python原价冲周董演唱会~Python自动秒杀抢票脚本
- 精通Python网络爬虫:核心技术、框架与项目实战.2.2 搜索引擎核心
- Python中python-nmap模块的使用
- python中pyqt5的进度条--python实战(十)
- 《树莓派Python编程入门与实战》——1.7 排除树莓派的故障
- 《树莓派Python编程入门与实战(第2版)》——1.3 哪些树莓派外围设备是必须的
- 《树莓派Python编程入门与实战(第2版)》——1.7 排除树莓派的故障
- 《数据科学实战手册(R+Python)》一第1章 准备你的数据科学环境
- 《Python数据可视化编程实战》—— 1.9 为项目设置matplotlib参数
- 《树莓派Python编程入门与实战》——1.6 让你的树莓派正常工作
- 《树莓派Python编程入门与实战》——1.9 Q&A
- Python实现softmax函数
- Python 数据分析教程之如何验证线性回归的假设,线性回归的假设是什么?以及如何用python验证它们?
- python 正则表达式删除HTML标签
- Jenkins持续集成实战之Jenkins构建Python项目提示:'python' 不是内部或外部命令,也不是可运行的程序。
- Python基础入门-函数实战登录功能
- linux:使用python脚本监控某个进程是否存在(不使用crontab)
- 学习笔记(43):Python实战编程-事件处理简介
- 学习笔记(40):Python实战编程-文本
- (原+转)简明 Python 教程:总结
- Python网络爬虫反爬破解策略实战