"<div class=\"toc\"><ul class=\"toc-item\"><li><span><a href=\"#Thread-Versus-Coroutine:-A-Comparison\" data-toc-modified-id=\"Thread-Versus-Coroutine:-A-Comparison-1\"><span class=\"toc-item-num\">1&nbsp;&nbsp;</span>Thread Versus Coroutine: A Comparison</a></span></li><li><span><a href=\"#download_flag-with-asyncio\" data-toc-modified-id=\"download_flag-with-asyncio-2\"><span class=\"toc-item-num\">2&nbsp;&nbsp;</span>download_flag with asyncio</a></span><ul class=\"toc-item\"><li><span><a href=\"#downlaod_flags2:-增加-error-handling\" data-toc-modified-id=\"downlaod_flags2:-增加-error-handling-2.1\"><span class=\"toc-item-num\">2.1&nbsp;&nbsp;</span>downlaod_flags2: 增加 error handling</a></span></li><li><span><a href=\"#download_flags3:-处理多个请求\" data-toc-modified-id=\"download_flags3:-处理多个请求-2.2\"><span class=\"toc-item-num\">2.2&nbsp;&nbsp;</span>download_flags3: 处理多个请求</a></span></li></ul></li><li><span><a href=\"#用-asyncio-写服务器\" data-toc-modified-id=\"用-asyncio-写服务器-3\"><span class=\"toc-item-num\">3&nbsp;&nbsp;</span>用 asyncio 写服务器</a></span></li></ul></div>"

{

"cell_type": "markdown",

"metadata": {},

"source": [

"<center>重点:asyncio.Future 与 concurrent.future.Future 的不同</center>\n",

"\n",

"1. futures 对象是 results of scheduling something for execution. 在 asyncio 中, BaseEventLoop.creat_task() 接收一个协程函数, 调度它运行, 并且立刻返回一个 asyncio.Task 实例, Task 是 Future 的子类.\n",

"2. 在 asyncio.Future 的 result() 方法没有参数, 你不能指定 timeout, 并且如果 future 没有完成, 它不会像 concurrent 中的 Future 那样阻塞主线程, 而是抛出\n",

" asyncio.InvalidStateError 异常.\n",

"3. 但是如果使用 yield from 语句而不是 result() 方法, 那么 future 会自动等待完成, 并将结果赋值给 yield from 左边的表达式. **事件循环**不会阻塞; 运行 yield from 的 协程函数会挂起, 直到 yield from 完成.\n",

"\n",

"有两种方法来创建 Task 对象:\n",

"\n",

"1. asyncio.async(coro_or_future, * , loop=None). 如果你传递了协程对象, async 将调用 loop.create_task() 来创建 Task 对象. 你亦可以指定事件循环, 如果不指定, 会调用 asyncio.get_event_loop() 获得**唯一的**事件循环.\n",

"2. BaseEventLoop.creat_task(coro) 调度协程函数并且立刻返回一个 Task 对象."

]

},

{

"cell_type": "markdown",

"metadata": {},

"source": [

"##### download_flag with asyncio\n",

"下面给出 donwload_flag 的 asyncio 版本, 注意:\n",

"\n",

"1. 需要 python 3.7\n",

"2. 无法在 jupyter notebook 中运行, 需要在命令行下运行"

]

},

{

"cell_type": "code",

"execution_count": 3,

"metadata": {

"ExecuteTime": {

"end_time": "2019-08-06T06:19:22.903533Z",

"start_time": "2019-08-06T06:19:22.900072Z"

}

},

"outputs": [],

"source": [

"import os\n",

"import time\n",

"import sys\n",

"\n",

"import asyncio # 1\n",

"import aiohttp # 2\n",

"\n",

"POP20_CC = ('CN IN US ID BR PK NG BD RU JP '\n",

" 'MX PH VN ET EG DE IR TR CD FR').split()\n",

"BASE_URL = 'http://flupy.org/data/flags'\n",

"DEST_DIR = 'downloads/'\n",

"\n",

"\n",

"def save_flag(img, filename):\n",

" path = os.path.join(DEST_DIR, filename)\n",

" with open(path, 'wb') as fp:\n",

" fp.write(img)\n",

"\n",

"\n",

"async def get_flag(session, cc): # 3 async 关键字 等同于 @asyncio.coroutine\n",

" url = '{}/{cc}/{cc}.gif'.format(BASE_URL, cc=cc.lower())\n",

" # 4 需要在 with 前加上 async, 因为 session.get是 IO bound 的,\n",

" async with session.get(url) as resp:\n",

" # 这一句等价于 resp = yield from aiohttp.request('GET', url)\n",

" return await resp.read() # 5 等价于 return yield from resp.read()\n",

"\n",

"\n",

"def show(text):\n",

" print(text, end=' ')\n",

" sys.stdout.flush()\n",

"\n",

"\n",

"async def download_one(session, cc): # 6 download_one 也是一个协程函数, 因为它调用了 get_flag 协程\n",

" # 7 等价于 yield from get_flag(session, cc)\n",

" image = await get_flag(session, cc)\n",

" show(cc)\n",

" save_flag(image, cc.lower() + '.gif')\n",

" return cc\n",

"\n",

"\n",

"async def download_many(cc_list):\n",

" async with aiohttp.ClientSeesion() as session: # 8 aiohttp 的 异步 session\n",

" res = await asyncio.gather( # 9 你无须显示的调用事件循环了.\n",

" *[asyncio.creat_task(download_one(session, cc)) for cc in sorted(cc_list)]\n",

" )\n",

" return len(res)\n",

"\n",

"\n",

"def main():\n",

" t0 = time.time()\n",

" count = asyncio.run(download_many(POP20_CC))\n",

" elapsed = time.time() - t0\n",

" msg = '\\n{} flags downloaded in {:.2f}s'\n",

" print(msg.format(count, elapsed))"

]

},

{

"cell_type": "code",

"execution_count": 14,

"metadata": {

"ExecuteTime": {

"end_time": "2019-08-06T06:43:39.003691Z",

"start_time": "2019-08-06T06:43:39.001286Z"

}

},

"outputs": [],

"source": [

"# main()"

]

},

{

"cell_type": "markdown",

"metadata": {},

"source": [

"对比一下老版本的 asyncio 实现, 注意:\n",

"\n",

"1. 仍然需要在命令行下实现\n",

"2. asyncio.wait 函数返回一个迭代器. wait 函数可以指定 **timeout 和 return_when** ;迭代器作为 loop.run_until_complete的参数后, run_until_complete\n",

" 将返回一个元组: (已完成的futures, 未完成的futures)---默认行为是等待所有 futures 完成"

]

},

{

"cell_type": "code",

"execution_count": 17,

"metadata": {

"ExecuteTime": {

"end_time": "2019-08-06T07:27:48.581705Z",

"start_time": "2019-08-06T07:27:48.574172Z"

}

},

"outputs": [],

"source": [

"import asyncio, aiohttp\n",

"\n",

"@asyncio.coroutine\n",

"def get_flag(cc):\n",

" url = '{}/{cc}/{cc}.gif'.format(BASE_URL, cc=cc.lower())\n",

" resp = yield from aiohttp.request('GET', url)\n",

" image = yield from resp.read()\n",

" return image\n",

"\n",

"@asyncio.coroutine\n",

"def download_one(cc):\n",

" image = yield from get_flag(cc)\n",

" show(cc)\n",

" save_flag(image, cc.lower() + '.gif')\n",

" return cc\n",

"\n",

"def download_many(cc_list):\n",

" \"\"\"这一版本的download_many需要显式地处理事件循环\n",

" \"\"\"\n",

" loop = asyncio.get_event_loop() #1 创建当前线程事件循环的引用\n",

" to_do = [download_one(cc) for cc in sorted(cc_list)] #2 创建生成器函数的list\n",

" wait_coro = asyncio.wait(to_do) #3 这不是一个阻塞函数. 他是一个协程函数, 接受可迭代的对象, 可迭代的对象里储存了 futures或协程函数 ,\n",

" # 当这些协程函数完成时, 它也完成.\n",

" res, _ = loop.run_until_complete(wait_coro) #4 执行事件循环直到 wait_coro 终止\n",

" loop.close() #5 关闭事件循环\n",

"\n",

"def main(download_many): \n",

" t0 = time.time()\n",

" count = download_many(POP20_CC)\n",

" elapsed = time.time() - t0\n",

" msg = '\\n{} flags downloaded in {:.2f}s'\n",

" print(msg.format(count, elapsed))"

]

},

{

"cell_type": "markdown",

"metadata": {},

"source": [

"<center>总结</center>\n",

"\n",

"1. yield from 会使得当前协程挂起, 控制流回到事件循环.\n",

"2. 生成器函数调用链的顶层必须是一个非协程的调用函数. 在 asyncio 里, 它调用了 loop.run_until_complete(); 无需显示调用 next 和 send.\n",

"3. 生成器函数调用链的底层必须是一个最简单的协程函数. 在 asyncio 里, 他要么是 asyncio 函数(比如 asyncio.sleep) 或者是 第三方库提供的协程函数(比如 aiohttp.request 和 resp.read()), 这些函数执行了 IO 操作.\n",

"4. save\\_flag 操作也是IO-bound 的, 但是当时没有提供异步文件操作的API."

]

},

{

"cell_type": "markdown",

"metadata": {},

"source": [

"###### downlaod_flags2: 增加 error handling\n",

"请查看 flags2_\\*.py 系列代码"

]

},

{

"cell_type": "markdown",

"metadata": {},

"source": [

"###### download_flags3: 处理多个请求\n",

"现在你不仅仅需要保存国旗图片, 你还需要保存相应的contry code. 因此在 get\\_flag 函数中需要做两件事:\n",

"\n",

"1. 获得国旗图片\n",

"2. 获得 metadata.json 文件\n",

"\n",

"下面实现了多个请求的异步并发的代码"

]

},

{

"cell_type": "code",

"execution_count": 2,

"metadata": {

"ExecuteTime": {

"end_time": "2019-08-09T04:03:45.578381Z",

"start_time": "2019-08-09T04:03:45.566879Z"

}

},

"outputs": [],

"source": [

"import asyncio\n",

"import aiohttp\n",

"from aiohttp import web\n",

"\n",

"\n",

"async def http_get(url):\n",

" \"\"\"处理 http 请求的协程函数.\n",

" 如果 response 的 status 是 200, 那么根据 response\n",

" 的 header 来判断 返回的是 metadata 还是 image\n",

" \"\"\"\n",

" res = await aiohttp.request('GET', url)\n",

" if res.status == 200:\n",

" ctype = res.headers.get('Content-type', '').lower()\n",

" if 'json' in ctype or url.endswith('json'):\n",

" data = await res.json() # 1\n",

" else:\n",

" data = await res.head() # 2\n",

" return data\n",

" elif res.status == 404:\n",

" raise web.HTTPNotFound()\n",

" else:\n",

" raise aiohttp.errors.HttpProcessingError(\n",

" code=res.status,\n",

" message=res.reason,\n",

" headers=res.headers)\n",

"\n",

"\n",

"async def get_country(base_url, cc):\n",

" url = '{}/{cc}/metadata.json'.format(base_url, cc=cc.lower())\n",

" metadata = await http_get(url) # 3\n",

" return metadata['country']\n",

"\n",

"\n",

"async def get_flag():\n",

" url = '{}/{cc}/{cc}.gif'.format(base_url, cc=cc.lower())\n",

" return (await http_get(url)) \n",

"\n",

"\n",

"async def download_one(cc, base_url, semaphore, verbose):\n",

" try:\n",

" async with semaphore: # 4\n",

" image = await get_flag(base_url, cc)\n",

" async with semaphore:\n",

" country = await get_country(base_url, cc)\n",

" except web.HTTPNotFound:\n",

" status = HTTPStatus.not_found\n",

" msg = 'not found'\n",

" except Exception as exc:\n",

" raise FetchError(cc) from exc\n",

" else:\n",

" country = country.replace(' ', '_')\n",

" filename = '{}-{}.gif'.format(country, cc)\n",

" loop = asyncio.get_event_loop()\n",

" loop.run_in_executor(None, save_flag, image, filename)\n",

" status = HTTPStatus.ok\n",

" msg = 'OK'\n",

" if verbose and msg:\n",

" print(cc, msg)\n",

" return Result(status, cc)"

]

},

{

"cell_type": "markdown",

"metadata": {},

"source": [

"<center>总结</center>\n",

"\n",

"1. 如果 `content type` 里有 'json', 或者 `url` 参数以 '.json' 结尾, 那么 yield from response.json(), 返回一个字典.\n",

"2. 否则, yield from response.read().\n",

"3. metadata 得到一个 json dict.\n",

"4. 将 get\\_flag 和 get\\_country 写在两个 semaphore 的块下, 这样每次 acquire semaphore 的时间尽量短."

]

},

{

"cell_type": "markdown",

"metadata": {},

"source": [

"##### 用 asyncio 写服务器"

]

},