這篇文章主要介紹了提升python處理速度原理及方法實(shí)例,文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下

　導(dǎo)讀：作為日常生產(chǎn)開(kāi)發(fā)中非常實(shí)用的一門(mén)語(yǔ)言，python廣泛應(yīng)用于網(wǎng)絡(luò)爬蟲(chóng)、web開(kāi)發(fā)、自動(dòng)化測(cè)試、數(shù)據(jù)分析和人工智能等領(lǐng)域。但python是單線(xiàn)程的，想要提升python的處理速度，涉及到一個(gè)很關(guān)鍵的技術(shù)——協(xié)程。本篇文章，將講述python協(xié)程的理解與使用。

1、操作系統(tǒng)相關(guān)概念

　　在理解與使用協(xié)程之前，先簡(jiǎn)單的了解幾個(gè)與操作系統(tǒng)相關(guān)的概念，包括進(jìn)程、線(xiàn)程、同步和異步、阻塞與非阻塞。了解這些概念，對(duì)你學(xué)習(xí)協(xié)程、消息隊(duì)列、緩存等知識(shí)都有一定的幫助。

（1）進(jìn)程：

　　進(jìn)程是操作系統(tǒng)分配資源的最小單位，系統(tǒng)由一個(gè)個(gè)程序（進(jìn)程）組成的，一般而言，分為文本區(qū)域、數(shù)據(jù)區(qū)域和堆棧區(qū)域

　　文本區(qū)域存儲(chǔ)處理器執(zhí)行的代碼（機(jī)器碼），通常來(lái)說(shuō)，這是一個(gè)只讀區(qū)域，防止運(yùn)行的程序被意外的修改

　　數(shù)據(jù)區(qū)域存儲(chǔ)所有的變量和動(dòng)態(tài)分配的內(nèi)存，又細(xì)分為初始化的數(shù)據(jù)區(qū)（所有初始化的全局、靜態(tài)、常量以及外部變量）和未初始化的數(shù)據(jù)區(qū)（初始化未0的全局變量和靜態(tài)變量），初始化的變量最初保存在文本區(qū)，程序啟動(dòng)后被拷貝到初始化的數(shù)據(jù)區(qū)

　　堆棧區(qū)域存儲(chǔ)著活動(dòng)過(guò)程調(diào)用的指令和本地變量，在地址空間里，棧區(qū)緊連著堆區(qū)，他們的增長(zhǎng)方向相反，內(nèi)存是線(xiàn)性的，所以我們的代碼放在低地址的地方，由低向高增長(zhǎng)，棧區(qū)大小不可預(yù)測(cè)，隨開(kāi)隨用，因此放在高地址的地方，由高向低增長(zhǎng)。當(dāng)堆與棧指針重合的時(shí)候，意味著內(nèi)存耗盡，造成內(nèi)存溢出。

　　進(jìn)程的創(chuàng)建和銷(xiāo)毀都非常的消耗系統(tǒng)資源，是一種比較昂貴的操作。進(jìn)程為了自身能夠得到運(yùn)行，必須搶占式的爭(zhēng)奪CPU。對(duì)于單核CPU而言，在同一時(shí)間內(nèi)只能執(zhí)行一個(gè)進(jìn)程的代碼，所以在單核CPU上實(shí)現(xiàn)多進(jìn)程，是通過(guò)CPU的快速切換不同進(jìn)程來(lái)實(shí)現(xiàn)的，看上去就像是多個(gè)進(jìn)程同時(shí)執(zhí)行。

　　由于進(jìn)程間是隔離的，各自擁有自己的內(nèi)存資源，相比于線(xiàn)程的共享內(nèi)存而言，要更安全，不同進(jìn)程之間的數(shù)據(jù)只能通過(guò)IPC（Inter-Process Communication）進(jìn)行通信共享

（2）線(xiàn)程

　　線(xiàn)程是CPU調(diào)度的基本單位。如果進(jìn)程是一個(gè)容器，線(xiàn)程就是運(yùn)行在容器里面的程序，線(xiàn)程是屬于進(jìn)程的，同個(gè)進(jìn)程的多個(gè)線(xiàn)程共享進(jìn)程的內(nèi)存地址空間

　　線(xiàn)程間可以直接通過(guò)全局變量進(jìn)行通信，所以相對(duì)來(lái)說(shuō)，線(xiàn)程間通信是不太安全的，因此引入各種鎖的場(chǎng)景，這里將不闡述

　　當(dāng)一個(gè)線(xiàn)程奔潰了，會(huì)導(dǎo)致整個(gè)進(jìn)程也奔潰，即其它線(xiàn)程也掛了。這一點(diǎn)與進(jìn)程不一樣，一個(gè)進(jìn)程掛了，其他進(jìn)程照樣執(zhí)行

　　在多核操作系統(tǒng)中，默認(rèn)一個(gè)進(jìn)程內(nèi)只有一個(gè)線(xiàn)程，所以對(duì)多進(jìn)程處理就像是一個(gè)進(jìn)程一個(gè)核心

（3）同步和異步

　　同步和異步關(guān)注的是消息通信機(jī)制，所謂同步，就是在發(fā)出一個(gè)函數(shù)調(diào)用時(shí)，在沒(méi)有得到結(jié)果之前，該調(diào)用不會(huì)返回。一旦調(diào)用返回，就立即得到調(diào)用的返回值，即調(diào)用者主動(dòng)等待調(diào)用結(jié)果

　　所謂異步，就是在請(qǐng)求發(fā)出去后，這個(gè)調(diào)用就立即返回，但沒(méi)有返回結(jié)果，通過(guò)回調(diào)的方式告知該調(diào)用的實(shí)際結(jié)果

　　同步的請(qǐng)求，需要主動(dòng)讀寫(xiě)數(shù)據(jù)，并且等待結(jié)果；異步的請(qǐng)求，調(diào)用者不會(huì)立即得到結(jié)果。而是在調(diào)用發(fā)出后，被調(diào)用者通過(guò)狀態(tài)、通知來(lái)告訴調(diào)用者，或通過(guò)回調(diào)函數(shù)處理這個(gè)調(diào)用

（4）阻塞與非阻塞

　　阻塞和非阻塞關(guān)注的是程序在等待調(diào)用結(jié)果（消息，返回值）時(shí)的狀態(tài)

　　阻塞調(diào)用是指調(diào)用結(jié)果返回之前，當(dāng)前線(xiàn)程會(huì)被掛起。調(diào)用線(xiàn)程只有在得到結(jié)果之后才會(huì)返回

　　非阻塞調(diào)用指在得到不能立即得到結(jié)果之前，該調(diào)用不會(huì)阻塞當(dāng)前線(xiàn)程。所以，區(qū)分的條件在于，進(jìn)程/線(xiàn)程要訪(fǎng)問(wèn)的數(shù)據(jù)是否就緒，進(jìn)程/線(xiàn)程是否需要等待

　　非阻塞一般通過(guò)多路復(fù)用實(shí)現(xiàn)，多路復(fù)用由select、poll、epoll幾種實(shí)現(xiàn)方式

（5）協(xié)程

　　了解完前面幾個(gè)概念，再來(lái)看看協(xié)程的概念

　　協(xié)程是屬于線(xiàn)程的，又稱(chēng)微線(xiàn)程，纖程，英文名是coroutine。舉個(gè)例子，在執(zhí)行函數(shù)A時(shí)，我希望能隨時(shí)終端去執(zhí)行函數(shù)B，然后終端B的執(zhí)行，切換回來(lái)執(zhí)行函數(shù)A。這就是協(xié)程的作用，由調(diào)用者自有切換。這個(gè)切換過(guò)程并不等同于函數(shù)調(diào)用，因?yàn)樗鼪](méi)有調(diào)用語(yǔ)句。執(zhí)行方式與多線(xiàn)程類(lèi)似，但是協(xié)程只有一個(gè)線(xiàn)程執(zhí)行

　　協(xié)程的優(yōu)點(diǎn)是執(zhí)行效率非常高，因?yàn)閰f(xié)程的切換是由程序自身控制，不需要切換線(xiàn)程，即沒(méi)有切換線(xiàn)程的開(kāi)銷(xiāo)。同時(shí)，由于只有一個(gè)線(xiàn)程，不存在沖突的問(wèn)題，不需要依賴(lài)鎖（加鎖和釋放鎖需要很多資源消耗）

　　協(xié)程的主要使用場(chǎng)景在于處理io密集型程序，解決效率問(wèn)題，不同于CPU密集型程序的處理。然而實(shí)際開(kāi)發(fā)中這兩種場(chǎng)景非常多，如果要充分發(fā)揮CPU的利用率，可以使用多進(jìn)程+協(xié)程的方式，本文后續(xù)將講到結(jié)合點(diǎn)

2、協(xié)程相關(guān)原理

　　根據(jù)wikipedia的定義，協(xié)程是一個(gè)無(wú)優(yōu)先級(jí)的子程序調(diào)度組件，允許子程序在特定的地方掛起恢復(fù)。所以理論上，只要內(nèi)存足夠，一個(gè)線(xiàn)程可以有任意多個(gè)協(xié)程，但同一時(shí)刻只能有一個(gè)協(xié)程在運(yùn)行，多個(gè)協(xié)程分享該線(xiàn)程分配到的計(jì)算機(jī)資源。協(xié)程是為了充分發(fā)揮異步調(diào)用的優(yōu)勢(shì)，異步操作則是為了IO操作阻塞線(xiàn)程

（1）知識(shí)準(zhǔn)備

　　在了解原理前，先做一個(gè)知識(shí)的準(zhǔn)備

　　1）現(xiàn)代主流的操作系統(tǒng)幾乎都是分時(shí)操作系統(tǒng)，即一臺(tái)計(jì)算機(jī)采用時(shí)間片輪轉(zhuǎn)的方式為多個(gè)用戶(hù)提供服務(wù)，系統(tǒng)資源分配的基本單位是進(jìn)程，CPU調(diào)度的基本單位是線(xiàn)程

　　2）運(yùn)行時(shí)內(nèi)存空間氛圍變量區(qū)、棧區(qū)、堆區(qū)。內(nèi)存地址分配上，堆區(qū)從低到高，棧區(qū)從高到低

　　3）計(jì)算機(jī)執(zhí)行時(shí)一條條指令讀取執(zhí)行，執(zhí)行到當(dāng)前指令時(shí)，下一條指令的指令的地址在指令寄存器的IP中，ESP寄存值只想當(dāng)前棧頂?shù)刂罚珽BP指向當(dāng)前活動(dòng)棧幀的基地址

　　4）系統(tǒng)發(fā)生函數(shù)調(diào)用時(shí)操作為：先將入?yún)挠彝笠淮螇簵＃缓蟀逊祷氐刂穳簵?，最后將?dāng)前EBP寄存器的值壓棧，修改ESP寄存器的值，在棧區(qū)分配當(dāng)前函數(shù)局部變量所需的空間

　　5）協(xié)程的上下文包含屬于當(dāng)前協(xié)程的棧區(qū)和寄存器里面存放的值

（2）事件循環(huán)

　　在python3.3中通過(guò)yield from使用協(xié)程，在3.5中，引入了關(guān)于協(xié)程的語(yǔ)法糖async/await的原理解析。其中，事件循環(huán)是一個(gè)核心所在，編寫(xiě)過(guò)js的同學(xué)，會(huì)對(duì)事件循環(huán)Eventloop更加了解，事件循環(huán)是一種等待程序分配消息或事件的編程架構(gòu)。在python中，asyncio.coroutine修飾器用來(lái)標(biāo)記作為協(xié)程的函數(shù)，這里的協(xié)程是和asyncio及其事件循環(huán)一起使用的，而在后續(xù)的發(fā)展中，async/await被使用的越來(lái)越廣泛

（3）async/await

　　async/await是使用python協(xié)程的關(guān)鍵，從結(jié)構(gòu)上來(lái)看，asyncio實(shí)質(zhì)上是一個(gè)異步框架，async/await是為異步框架提供API以方便使用者調(diào)用，所以使用者要想使用async/await編寫(xiě)協(xié)程代碼，目前必須基于asyncio或其他異步庫(kù)

（4）Future

　　在實(shí)際開(kāi)發(fā)編寫(xiě)異步代碼時(shí)，為了避免太多回調(diào)方法導(dǎo)致的回調(diào)地獄，但又需要獲取異步調(diào)用的返回結(jié)果，聰明的語(yǔ)言設(shè)計(jì)者設(shè)計(jì)了一個(gè)叫做Future的對(duì)象，封裝了與loop的交互行為。其大致執(zhí)行過(guò)程為：程序啟動(dòng)后，通過(guò)add_done_callback方法向epoll注冊(cè)回調(diào)函數(shù)，當(dāng)result屬性得到返回值后，主動(dòng)運(yùn)行之前注冊(cè)的回調(diào)函數(shù)，向上傳遞給coroutine。這個(gè)Future對(duì)象為asyncio.Future

　　但是，要想取得返回值，程序必須恢復(fù)到工作狀態(tài)，而由于Future對(duì)象本身的生存周期比較短，每一次注冊(cè)回調(diào)、產(chǎn)生事件、觸發(fā)回調(diào)過(guò)程后工作可能已經(jīng)完成，所以用Future向生成器send result并不合適。這里又引入一個(gè)新的對(duì)象Task，保存在Future對(duì)象中，對(duì)生成器協(xié)程進(jìn)行狀態(tài)管理

　　Python里另一個(gè)Future對(duì)象是concurrent.futures.Future，與asyncio.Future互不兼容，容易產(chǎn)生混淆。區(qū)別點(diǎn)在于，concurrent.futures是線(xiàn)程級(jí)的Future對(duì)象，當(dāng)使用concurrent.futures.Executor進(jìn)行多線(xiàn)程編程時(shí)，該對(duì)象用于在不同的thread之間傳遞結(jié)果

（5）Task

　　上文中提到，Task是維護(hù)生成器協(xié)程狀態(tài)處理執(zhí)行邏輯的任務(wù)對(duì)象，Task中有一個(gè)_step方法，負(fù)責(zé)生成器協(xié)程與EventLoop交互過(guò)程的狀態(tài)遷移，整個(gè)過(guò)程可以理解為：Task向協(xié)程send一個(gè)值，恢復(fù)其工作狀態(tài)。當(dāng)協(xié)程運(yùn)行到斷點(diǎn)后，得到新的Future對(duì)象，再處理future與loop的回調(diào)注冊(cè)過(guò)程

（6）Loop

　　在日常開(kāi)發(fā)中，會(huì)有一個(gè)誤區(qū)，認(rèn)為每一個(gè)線(xiàn)程都可以有一個(gè)獨(dú)立的loop。實(shí)際運(yùn)行時(shí)，主線(xiàn)程才能通過(guò)asyncio.get_event_loop()創(chuàng)建一個(gè)新的loop，而在其他線(xiàn)程時(shí)，使用get_event_loop()卻會(huì)拋錯(cuò)。正確的做法為通過(guò)asyncio.set_event_loop()，將當(dāng)前線(xiàn)程與主線(xiàn)程loop顯式綁定

3、協(xié)程實(shí)戰(zhàn)

　　上面介紹完了協(xié)程相關(guān)的概念和原理，接下來(lái)看看如何使用，這里舉一個(gè)實(shí)際場(chǎng)景的例子

場(chǎng)景：
　　外部接受一些文件，每個(gè)文件里有一些數(shù)據(jù)，其中，這組數(shù)據(jù)需要通過(guò)http的方式，發(fā)向第三方平臺(tái)，并獲得結(jié)果

分析：
　　由于同一文件的每一組數(shù)據(jù)沒(méi)有前后的處理邏輯，在之前通過(guò)requests庫(kù)發(fā)送的網(wǎng)絡(luò)請(qǐng)求，串行執(zhí)行，下一組數(shù)據(jù)的發(fā)送需要等待上一組數(shù)據(jù)的返回，顯得整個(gè)文件的處理時(shí)間長(zhǎng)，這種請(qǐng)求方式，完全可以由協(xié)程來(lái)實(shí)現(xiàn)

　　為了更方便的配合協(xié)程發(fā)請(qǐng)求，我們使用aiohttp庫(kù)來(lái)代替requests庫(kù)，關(guān)于aiohttp，下面做簡(jiǎn)單介紹

aiohttp：

　　aiohttp是asyncio和python的異步HTTP客戶(hù)端/服務(wù)器，由于是異步的，經(jīng)常用在服務(wù)器端接收請(qǐng)求，和客戶(hù)端爬蟲(chóng)應(yīng)用，發(fā)起異步請(qǐng)求，這里我們主要用來(lái)發(fā)請(qǐng)求

　　aiohttp支持客戶(hù)端和HTTP服務(wù)器，可以實(shí)現(xiàn)單線(xiàn)程并發(fā)IO操作，無(wú)需使用Callback Hell即可支持Server WebSockets和Client WebSockets，且具有中間件

4、代碼實(shí)現(xiàn)

直接上代碼吧，talk is cheap,show me the code~

import aiohttp
import asyncio
from inspect import isfunction
import time
import logger

@logging_utils.exception(logger)
def request(pool, data_list):
  loop = asyncio.get_event_loop()
  loop.run_until_complete(exec(pool, data_list))
async def exec(pool, data_list):
  tasks = []
  sem = asyncio.Semaphore(pool)
  for item in data_list:
    tasks.append(
      control_sem(sem,
            item.get("method", "GET"),
            item.get("url"),
            item.get("data"),
            item.get("headers"),
            item.get("callback")))
  await asyncio.wait(tasks)
async def control_sem(sem, method, url, data, headers, callback):
  async with sem:
    count = 0
    flag = False
    while not flag and count < 4:
      flag = await fetch(method, url, data, headers, callback)
      count = count + 1
      print("flag:{},count:{}".format(flag, count))
    if count == 4 and not flag:
      raise Exception('EAS service not responding after 4 times of retry.')
async def fetch(method, url, data, headers, callback):
  async with aiohttp.request(method, url=url, data=data, headers=headers) as resp:
    try:
      json = await resp.read()
      print(json)
      if resp.status != 200:
        return False
      if isfunction(callback):
        callback(json)
      return True
    except Exception as e:
      print(e)

這里，我們封裝了對(duì)外發(fā)送批量請(qǐng)求的request方法，接收一次性發(fā)送的數(shù)據(jù)多少，和數(shù)據(jù)綜合，在外部使用時(shí)，只需要構(gòu)建好網(wǎng)絡(luò)請(qǐng)求對(duì)象的數(shù)據(jù)，設(shè)定好請(qǐng)求池大小即可，同時(shí)，設(shè)置了重試功能，進(jìn)行了4次重試，防治在網(wǎng)絡(luò)抖動(dòng)的時(shí)候，單個(gè)數(shù)據(jù)的網(wǎng)絡(luò)請(qǐng)求發(fā)送失敗

最終效果：

在使用協(xié)程重構(gòu)網(wǎng)絡(luò)請(qǐng)求模塊之后，當(dāng)數(shù)據(jù)量在1000的時(shí)候，由之前的816s，提升到424s，快了一倍，且請(qǐng)求池大小加大的時(shí)候，效果更明顯，由于第三方平臺(tái)同時(shí)建立連接的數(shù)據(jù)限制，我們?cè)O(shè)定了40的閾值。可以看到，優(yōu)化的程度很顯著

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

最新評(píng)論