银行网站建设公司,大连华南网站建设,淘宝客做自已的网站,大庆市建设局网站上不去文章目录1. 线程与协程对比2. 使用 asyncio 和 aiohttp 下载3. 避免阻塞型调用4. 使用 asyncio.as_completed5. 使用Executor对象#xff0c;防止阻塞事件循环6. 从回调到期物和协程learn from 《流畅的python》
1. 线程与协程对比
threading
import threading
import iter…
文章目录1. 线程与协程对比2. 使用 asyncio 和 aiohttp 下载3. 避免阻塞型调用4. 使用 asyncio.as_completed5. 使用Executor对象防止阻塞事件循环6. 从回调到期物和协程learn from 《流畅的python》
1. 线程与协程对比
threading
import threading
import itertools
import time
import sysclass Signal:go Truedef spin(msg, signal):write, flush sys.stdout.write, sys.stdout.flushfor char in itertools.cycle(|/-\\): # 无限循环status char msgwrite(status)flush()write(\x08 * len(status)) # \x08 退格键光标移动回去time.sleep(0.1)if not signal.go:breakwrite( * len(status) \x08 * len(status))# 使用空格清除状态消息把光标移回开头def slow_function(): # 假设是一个耗时的计算过程time.sleep(10) # sleep 会阻塞主线程释放GIL创建从属线程return 42def supervisor(): # 该函数设置从属线程显示线程对象运行耗时的计算最后杀死线程signal Signal()spinner threading.Thread(targetspin, args(thinking!, signal))print(spinner object:, spinner) # 显示从属线程对象spinner.start() # 启动从属线程result slow_function() # 运行计算程序阻塞主线程从属线程动画显示旋转指针signal.go False # 改变signal 状态终止 spin 中的for循环spinner.join() # 等待spinner线程结束return resultdef main():result supervisor() # 运行 supervisorprint(Answer:, result)if __name__ __main__:main()适合 asyncio 的协程要由调用方驱动并由调用方通过 yield from 调用语法过时了新版的用 async / await 或者把协程传给 asyncio 包中的某个函数
一篇博文参考https://www.cnblogs.com/dhcn/p/9032461.html
import asyncio
import itertools
import sys# https://docs.python.org/3.8/library/asyncio.html
async def spin(msg): # py3.5以后的新语法 async / await协程函数write, flush sys.stdout.write, sys.stdout.flushfor char in itertools.cycle(|/-\\): # 无限循环status char msgwrite(status)flush()write(\x08 * len(status)) # \x08 退格键光标移动回去try:await asyncio.sleep(0.1)except asyncio.CancelledError: # 遇到取消异常退出循环print(cancel)breakwrite( * len(status) \x08 * len(status))print(end spin)async def slow_function(): # 协程函数print(start IO)await asyncio.sleep(3) # 假装进行 IO 操作print(end IO )return 42async def supervisor(): # 协程函数spinner asyncio.ensure_future(spin(thinking!)) # spinner 排定任务print(spinner object:, spinner) # 显示从属线程对象# spinner object: Task pending corospin() running at D:\ print(start slow)result await slow_function()print(end slow)spinner.cancel() # task对象可以取消抛出CancelledError异常return resultdef main():loop asyncio.get_event_loop() # 获取事件循环的引用result loop.run_until_complete(supervisor()) # 驱动 supervisor 协程让它运行完毕loop.close()print(answer:, result)if __name__ __main__:main()输出
spinner object: Task pending corospin() running at D:\gitcode
start slow
start IO
end IO ng!期间thinking在输出后来被覆盖
end slow
cancel
end spin
answer: 42请按任意键继续. . .2. 使用 asyncio 和 aiohttp 下载
import time
import sys
import os
import asyncio
import aiohttpPOP20_CC (CN IN US ID BR PK NG BD RU JP MX PH VN ET EG DE IR TR CD FR).split()
BASE_URL http://flupy.org/data/flags
DEST_DIR ./def save_flag(img, filename): # 保存图像path os.path.join(DEST_DIR, filename)with open(path, wb) as fp:fp.write(img)def show(text): # 打印信息print(text, end )sys.stdout.flush()async def get_flag(cc): # 获取图像url {}/{cc}/{cc}.gif.format(BASE_URL, cccc.lower())async with aiohttp.request(GET, url) as resp:image await resp.read()return imageasync def download_one(cc):image await get_flag(cc)show(cc)save_flag(image, cc.lower() .gif)return ccdef download_many_(cc_list):loop asyncio.get_event_loop()todo [download_one(cc) for cc in sorted(cc_list)] # 协程对象wait_coro asyncio.wait(todo) # 包装成 taskwait是协程函数返回协程或者生成器对象res, _ loop.run_until_complete(wait_coro)# 驱动协程返回 第一个元素是一系列结束的期物第二个元素是一系列未结束的期物# loop.close()好像不需要这句 上面 with 处可能自动关闭了return len(res)def main(download_many):t0 time.time()count download_many(POP20_CC)elapsed time.time() - t0msg \n{} flags downloaded in {:.2f}sprint(msg.format(count, elapsed)) # 计时信息if __name__ __main__:main(download_many_)# US RU ID ET BR FR CN PH BD NG DE JP EG TR MX IN PK IR CD VN
# 20 flags downloaded in 3.88s3. 避免阻塞型调用
执行硬盘或网络 I/O 操作的函数定义为 阻塞型函数 有两种方法能 避免阻塞型调用 中止整个应用程序 的进程
在单独的线程中运行各个阻塞型操作把每个阻塞型操作 转换成非阻塞的异步调用 使用
4. 使用 asyncio.as_completed
import collections
import time
import sys
import os
import asyncio
from http import HTTPStatusimport aiohttp
from aiohttp import web
import tqdmPOP20_CC (CN IN US ID BR PK NG BD RU JP MX PH VN ET EG DE IR TR CD FR).split()
BASE_URL http://flupy.org/data/flags
DEST_DIR ./
DEFAULT_CONCUR_REQ 5
MAX_CONCUR_REQ 1000class FetchError(Exception):def __init__(self, country_code):self.country_code country_codedef save_flag(img, filename): # 保存图像path os.path.join(DEST_DIR, filename)with open(path, wb) as fp:fp.write(img)def show(text): # 打印信息print(text, end )sys.stdout.flush()async def get_flag(cc): # 获取图像url {}/{cc}/{cc}.gif.format(BASE_URL, cccc.lower())async with aiohttp.request(GET, url) as resp:if resp.status 200:image await resp.read()return imageelif resp.status 404:raise web.HTTPNotFound()else:raise aiohttp.WebSocketError(coderesp.status, messageresp.reason)async def download_one(cc, semaphore, verbose):try:async with semaphore:image await get_flag(cc)except web.HTTPNotFound:status HTTPStatus.NOT_FOUNDmsg not foundexcept Exception as exc:raise FetchError(cc) from excelse:save_flag(image, cc.lower() .gif)status HTTPStatus.OKmsg OKif verbose and msg:print(cc, msg)return (status, cc)async def downloader_coro(cc_list, verbose, concur_req): # 协程函数counter collections.Counter()semaphore asyncio.Semaphore(valueconcur_req) # 最多可以使用这个计数器的协程个数todo [download_one(cc, semaphore, verboseTrue) for cc in sorted(cc_list)] # 协程对象列表todo_iter asyncio.as_completed(todo) # 获取迭代器会在期物运行结束后返回期物if not verbose:todo_iter tqdm.tqdm(todo_iter, totallen(cc_list)) # 迭代器传给tqdm显示进度条for future in todo_iter: # 迭代器运行结束的期物try:res await future # 获取期物对象的结果except FetchError as exc:country_code exc.country_codetry:error_msg exc.__cause__.args[0]except IndexError:error_msg exc.__cause__.__class__.__name__if verbose and error_msg:msg *** Error for {}: {}print(msg.format(country_code, error_msg))status HTTPStatus.errorelse:status res[0]counter[status] 1 # 记录结果return counter # 返回计数器def download_many_(cc_list, verbose, concur_req):loop asyncio.get_event_loop()coro downloader_coro(cc_list, verboseverbose, concur_reqconcur_req)# 实例化 downloader_coro协程然后通过 run_until_complete 方法把它传给事件循环counts loop.run_until_complete(coro)# loop.close() # 好像不需要这句 上面 with 处可能自动关闭了return countsdef main(download_many):t0 time.time()count download_many(POP20_CC, True, MAX_CONCUR_REQ)elapsed time.time() - t0msg \n{} flags downloaded in {:.2f}sprint(msg.format(count, elapsed)) # 计时信息if __name__ __main__:main(download_many_)5. 使用Executor对象防止阻塞事件循环
loop.run_in_executor 方法把阻塞的作业例如保存文件委托给线程池做
async def download_one(cc, semaphore, verbose):try:async with semaphore:image await get_flag(cc)except web.HTTPNotFound:status HTTPStatus.NOT_FOUNDmsg not foundexcept Exception as exc:raise FetchError(cc) from excelse:# 因此保存文件时整个应用程序都会冻结,为了避免使用下面方法loop asyncio.get_event_loop() # 获取事件循环对象的引用loop.run_in_executor(None, # 方法的第一个参数是 Executor 实例# 如果设为 None使用事件循环的默认 ThreadPoolExecutor 实例save_flag, image, cc.lower() .gif)# 余下的参数是可调用的对象以及可调用对象的位置参数status HTTPStatus.OKmsg OKif verbose and msg:print(cc, msg)return (status, cc)6. 从回调到期物和协程
如果一个操作需要依赖之前操作的结果那就得嵌套回调
def stage1(response1):request2 step1(response1)api_call2(request2, stage2)def stage2(response2):request3 step2(response2)api_call3(request3, stage3)def stage3(response3):tep3(response3)api_call1(request1, stage1)好的写法
async def three_stages(request1): response1 await api_call1(request1) # 第一步 request2 step1(response1) response2 await api_call2(request2) # 第二步 request3 step2(response2) response3 await api_call3(request3)# 第三步 step3(response3)
loop.create_task(three_stages(request1)) # 必须显式调度执行协程 必须使用 事件循环 显式排定 协程的执行时间
异步系统 能 避免用户级线程的开销这是它能比多线程系统管理更多并发连接的主要原因
