Python從使用線(xiàn)程到使用async/await的深入講解

前言

為了簡(jiǎn)化并更好地標(biāo)識(shí)異步IO，從Python 3.5開(kāi)始引入了新的語(yǔ)法async和await，可以讓coroutine的代碼更簡(jiǎn)潔易讀。

請(qǐng)注意，async和await是針對(duì)coroutine的新語(yǔ)法，要使用新的語(yǔ)法，只需要做兩步簡(jiǎn)單的替換：

• 把@asyncio.rotoutine替換為async；
• 把yield from替換為await。

async/await 是一種異步變成方法，還有兩種你可能聽(tīng)過(guò)，

     1. 回調(diào)

     2. Promise

（寫(xiě)過(guò) JavaScript 的肯定很熟悉了）

異步意味著任務(wù)不會(huì)阻塞，比如，如果我要下載一個(gè)比較忙的網(wǎng)絡(luò)資源，我的程序不需要一直等待下載完成，它可以在等待下載時(shí)繼續(xù)做其他事情。這與并行執(zhí)行多個(gè)操作不同。以下偽代碼比較容易理解：

# 慢方法
page = get_page_sync('some_page')

# 會(huì)阻塞整個(gè)程序的運(yùn)行
print(page)

有兩種方法可以改善上述的情況

（一）首先，讓我們?cè)囋囀褂镁€(xiàn)程。通過(guò)使用線(xiàn)程，我們可以將 get_page_sync 調(diào)用放到單獨(dú)的線(xiàn)程去執(zhí)行，這樣主線(xiàn)程 就可以繼續(xù)執(zhí)行其他操作。

# 將慢方法放到單獨(dú)的線(xiàn)程執(zhí)行
t = threading.thread(
 target = get_page_sync('some_page'，args=('some_page',))
)
t.run()

# 在線(xiàn)程運(yùn)行時(shí)執(zhí)行其他操作
do_something_else()

# 等待線(xiàn)程完執(zhí)行成
t.join()

線(xiàn)程有幾個(gè)優(yōu)缺點(diǎn)，主要的缺點(diǎn)是：

     1. 必須在改變共享數(shù)據(jù)前鎖定共享數(shù)據(jù)

     2. 只能通過(guò)傳遞給主線(xiàn)程消息來(lái)處理線(xiàn)程內(nèi)的異常

（二）現(xiàn)在我們?cè)囋嚨诙N中的 async/await，Python3.5 開(kāi)始支持的 async/await 方式，與第一種（線(xiàn)程）之間的主要區(qū)別在于，后者是操作系統(tǒng)內(nèi)核執(zhí)行上下文切換，而前者中我們自己控制。（上下文切換即，當(dāng)多個(gè)線(xiàn)程正在運(yùn)行時(shí)，內(nèi)核可能停止當(dāng)前進(jìn)程，使其進(jìn)入休眠狀態(tài)，并選擇不同的線(xiàn)程繼續(xù)執(zhí)行。這被稱(chēng)作搶占式多任務(wù)處理【Preemption】）

當(dāng)我們自己控制時(shí)，它被稱(chēng)作非搶占式或合作型多任務(wù)式，因?yàn)槭俏覀冏约禾幚砩舷挛那袚Q，所以我們需要一個(gè)調(diào)度程序，也叫做『事件循環(huán)』。此事件循環(huán)只循環(huán)遍歷等待中的調(diào)度，并運(yùn)行它的所有事件。每當(dāng)我們產(chǎn)生操作時(shí)，當(dāng)前任務(wù)會(huì)被添加到隊(duì)列中，且第一個(gè)任務(wù)（優(yōu)先級(jí)而非順序）從隊(duì)列中彈出并開(kāi)始執(zhí)行。例如，可以通過(guò)以下方式更改上述偽代碼：

async def print_page():
 page = await get_page_sync('some_page')
 print(page)

當(dāng)我們觸發(fā)上面的語(yǔ)句時(shí)，get_page_async 方法將非阻塞的獲取 some_page 還有 yield 句柄，這意味著我們的 print_page 函數(shù)將控制時(shí)間循環(huán) ，并且時(shí)間循環(huán)可以繼續(xù)執(zhí)行其他曹組，知道我們得到返回的響應(yīng)。

我們先將我們的線(xiàn)程代碼改造成這種語(yǔ)法。我們將使用 asyncio（Python 自帶的時(shí)間循環(huán)庫(kù)），并使用 aiohttp 包來(lái)執(zhí)行異步 http 請(qǐng)求。

我們將會(huì)創(chuàng)建一個(gè)名為 main 函數(shù)，它將成為我們異步代碼的入口。然后我們創(chuàng)建一個(gè)時(shí)間循環(huán)和一個(gè)「未來(lái)對(duì)象」。這個(gè)未來(lái)對(duì)象是對(duì)異步函數(shù)的抽象，它存儲(chǔ)了一些基本的屬性，比如它當(dāng)前的狀態(tài)（就像 Promise 一樣） 。然后我們將告訴我們的時(shí)間循環(huán)繼續(xù)運(yùn)行，知道這個(gè)「未來(lái)」完成。

loop = asyncio.get_event_loop()
future = asyncio.ensure_future(main())
loop.run_until_complete(future)

在我們的 main 方法中，我們將創(chuàng)建另一個(gè)未來(lái)任務(wù)列表，每個(gè)任務(wù)負(fù)責(zé)從某網(wǎng)站下載不同的桐鄉(xiāng)。我們這樣做是因?yàn)槊看蜗螺d都會(huì)發(fā)起網(wǎng)絡(luò)請(qǐng)求，在網(wǎng)絡(luò)請(qǐng)求時(shí)，我們可以運(yùn)行另一端代碼。創(chuàng)建任務(wù)列表后，我們可以通過(guò)調(diào)用等待整個(gè)列表執(zhí)行完成 asyncio.gather ，這就是它的實(shí)現(xiàn)：

async def main():
 tasks = []
 async with aiohttp.ClientSession() as session:
  for img in img_list:
   task = asyncio.ensure_future(download_img(img, session))
   task.append(task)
 await asyncio.gather(*tasks)

（這段代碼來(lái)的有點(diǎn)猛了）

最后一個(gè)我們要改的方法就是 download_img 了，我們僅僅需要替換 requests.get 調(diào)用為異步：

 i = 1
async def download_img(img, session):
 global i, bar
 
 # 獲取文件后綴
 file_ext = get_extention(img.link)
 # 拼接文件名
 file_name = img.id + file_ext

 resp = await session.get(img.link)
 with open(file_name, 'wb') as f:
  async for chunk in resp.content.iter_chunked(1024):
   f.write(chunk)

 bar.update(i)
 i += 1

要注意的一點(diǎn)是在更新 i 的時(shí)候不需要先鎖住它，這是因?yàn)槲覀兦懊嬲f(shuō)過(guò)，沒(méi)有代碼是同時(shí)執(zhí)行的，所以永遠(yuǎn)不可能出現(xiàn)競(jìng)態(tài)條件。

因?yàn)闆](méi)有鎖或者線(xiàn)程的開(kāi)銷(xiāo)，異步版本可能還會(huì)比多線(xiàn)程版本快一些。

這是完整代碼：

#! /usr/bin/env python

import os
import re
import sys

import aiohttp
import asyncio
import async_timeout

import progressbar

from imgurpython import ImgurClient

regex = re.compile(r'\.(\w+)$')
def get_extension(link):
 ext = regex.search(link).group()

 return ext

i = 1
async def download_img(img, session):
 global i, bar

 # get the file extension
 file_ext = get_extension(img.link)
 # create unique name by combining file id with its extension
 file_name = img.id + file_ext

 resp = await session.get(img.link)
 with open(file_name, 'wb') as f:
  async for chunk in resp.content.iter_chunked(1024):
   f.write(chunk)

 bar.update(i)
 i += 1

try:
 album_id = sys.argv[1]
except IndexError:
 raise Exception('Please specify an album id')

client_id = os.getenv('IMGUR_CLIENT_ID')
client_secret = os.getenv('IMGUR_CLIENT_SECRET')
client = ImgurClient(client_id, client_secret)

img_lst = client.get_album_images(album_id)
bar = progressbar.ProgressBar(max_value=len(img_lst))

async def main():
 tasks = []
 async with aiohttp.ClientSession() as session:
  for img in img_lst:
   task = asyncio.ensure_future(download_img(img, session))
   tasks.append(task)

  await asyncio.gather(*tasks)

loop = asyncio.get_event_loop()
future = asyncio.ensure_future(main())
loop.run_until_complete(future)

原文：https://medium.com/@exqu17/python-bits-moving-from-threads-to-async-await-741ec5124cdc

作者：https://medium.com/@exqu17?source=post_header_lockup

總結(jié)

以上就是這篇文章的全部?jī)?nèi)容了，希望本文的內(nèi)容對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，如果有疑問(wèn)大家可以留言交流，謝謝大家對(duì)億速云的支持。