建站
咨詢
售后
建站咨詢
新聞中心
1. 協(xié)程相關(guān)的概念
專注于為中小企業(yè)提供網(wǎng)站建設(shè)、做網(wǎng)站服務(wù),電腦端+手機(jī)端+微信端的三站合一,更高效的管理,為中小企業(yè)徽縣免費(fèi)做網(wǎng)站提供優(yōu)質(zhì)的服務(wù)。我們立足成都，凝聚了一批互聯(lián)網(wǎng)行業(yè)人才，有力地推動(dòng)了上千家企業(yè)的穩(wěn)健成長，幫助中小企業(yè)通過網(wǎng)站建設(shè)實(shí)現(xiàn)規(guī)模擴(kuò)充和轉(zhuǎn)變。
1.1 進(jìn)程和線程
進(jìn)程（Process）是應(yīng)用程序啟動(dòng)的實(shí)例，擁有代碼、數(shù)據(jù)和文件和獨(dú)立的內(nèi)存空間，是操作系統(tǒng)最小資源管理單元。每個(gè)進(jìn)程下面有一個(gè)或者多個(gè)線程（Thread），來負(fù)責(zé)執(zhí)行程序的計(jì)算，是最小的執(zhí)行單元。
重點(diǎn)是：操作系統(tǒng)會(huì)負(fù)責(zé)進(jìn)程的資源的分配；控制權(quán)主要在操作系統(tǒng)。另一方面，線程做為任務(wù)的執(zhí)行單元，有新建、可運(yùn)行runnable（調(diào)用start方法，進(jìn)入調(diào)度池，等待獲取cpu使用權(quán)）、運(yùn)行running（得到cpu使用權(quán)開始執(zhí)行程序） 阻塞blocked（放棄了cpu 使用權(quán)，再次等待） 死亡dead5中不同的狀態(tài)。線程的轉(zhuǎn)態(tài)也是由操作系統(tǒng)進(jìn)行控制。線程如果存在資源共享的情況下，就需要加鎖，比如生產(chǎn)者和消費(fèi)者模式，生產(chǎn)者生產(chǎn)數(shù)據(jù)多共享隊(duì)列，消費(fèi)者從共享隊(duì)列中消費(fèi)數(shù)據(jù)。
線程和進(jìn)程在得到和放棄cpu使用權(quán)時(shí)，cpu使用權(quán)的切換都需損耗性能，因?yàn)槟硞€(gè)線程為了能夠在再次獲得cpu使用權(quán)時(shí)能繼續(xù)執(zhí)行任務(wù)，必須記住上一次執(zhí)行的所有狀態(tài)。另外線程還有鎖的問題。
1.2 并行和并發(fā)
并行和并發(fā)，聽起來都像是同時(shí)執(zhí)行不同的任務(wù)。但是這個(gè)同時(shí)的含義是不一樣的。
- 并行：多核CPU才有可能真正的同時(shí)執(zhí)行，就是獨(dú)立的資源來完成不同的任務(wù)，沒有先后順序。
- 并發(fā)（concurrent）：是看上去的同時(shí)執(zhí)行，實(shí)際微觀層面是順序執(zhí)行，是操作系統(tǒng)對進(jìn)程的調(diào)度以及cpu的快速上下文切換，每個(gè)進(jìn)程執(zhí)行一會(huì)然后停下來，cpu資源切換到另一個(gè)進(jìn)程，只是切換的時(shí)間很短，看起來是多個(gè)任務(wù)同時(shí)在執(zhí)行。要實(shí)現(xiàn)大并發(fā)，需要把任務(wù)切成小的任務(wù)。
上面說的多核cpu可能同時(shí)執(zhí)行，這里的可能是和操作系統(tǒng)調(diào)度有關(guān)，如果操作系統(tǒng)調(diào)度到同一個(gè)cpu，那就需要cpu進(jìn)行上下文切換。當(dāng)然多核情況下，操作系統(tǒng)調(diào)度會(huì)盡可能考慮不同cpu。
這里的上下文切換可以理解為需要保留不同執(zhí)行任務(wù)的狀態(tài)和數(shù)據(jù)。所有的并發(fā)處理都有排隊(duì)等候，喚醒，執(zhí)行至少三個(gè)這樣的步驟
1.3 協(xié)程
我們知道線程的提出是為了能夠在多核cpu的情況下，達(dá)到并行的目的。而且線程的執(zhí)行完全是操作系統(tǒng)控制的。而協(xié)程（Coroutine）是線程下的，控制權(quán)在于用戶，本質(zhì)是為了能讓多組過程能不獨(dú)自占用完所有資源，在一個(gè)線程內(nèi)交叉執(zhí)行，達(dá)到高并發(fā)的目的。
協(xié)程的優(yōu)勢：
- 協(xié)程最大的優(yōu)勢就是協(xié)程極高的執(zhí)行效率。因?yàn)樽映绦蚯袚Q不是線程切換，而是由程序自身控制，因此，沒有線程切換的開銷，和多線程比，線程數(shù)量越多，協(xié)程的性能優(yōu)勢就越明顯
- 第二大優(yōu)勢就是不需要多線程的鎖機(jī)制，因?yàn)橹挥幸粋€(gè)線程，也不存在同時(shí)寫變量沖突，在協(xié)程中控制共享資源不加鎖，只需要判斷狀態(tài)就好了，所以執(zhí)行效率比多線程高很多。
協(xié)程和線程區(qū)別：
- 協(xié)程都沒參與多核CPU并行處理，協(xié)程是不并行
- 線程在多核處理器上是并行在單核處理器是受操作系統(tǒng)調(diào)度的
- 協(xié)程需要保留上一次調(diào)用的狀態(tài)
- 線程的狀態(tài)有操作系統(tǒng)來控制
我們姑且也過一遍這些文字上的概念，show your code的時(shí)候再聯(lián)系起來，就會(huì)更清晰的。
2. python中的線程
python中的線程由于歷史原因，即使在多核cpu的情況下并不能達(dá)真正的并行。這個(gè)原因就是全局解釋器鎖GIL（global interpreter lock），準(zhǔn)確的說GIL不是python的特性，而是cpython引入的一個(gè)概念。cpython解釋器在解析多線程時(shí)，會(huì)上GIL鎖，保證同一時(shí)刻只有一個(gè)線程獲取CPU使用權(quán)。
- 為什么需要GIL python中一切都是對象，Cpython中對象的回收，是通過對象的引用計(jì)數(shù)來判斷，當(dāng)對象的引用計(jì)數(shù)為0時(shí)，就會(huì)進(jìn)行垃圾回收，自動(dòng)釋放內(nèi)存。但是如果多線程的情況，引用計(jì)數(shù)就變成了一個(gè)共享的變量 Cpython是當(dāng)下最流行的Python的解釋器，使用引用計(jì)數(shù)來管理內(nèi)存，在Python中，一切都是對象，引用計(jì)數(shù)就是指向?qū)ο蟮闹羔様?shù)，當(dāng)這個(gè)數(shù)字變成0，則會(huì)進(jìn)行垃圾回收，自動(dòng)釋放內(nèi)存。但是問題是Cpython是線程不安全的。
考慮下如果有兩個(gè)線程A和B同時(shí)引用一個(gè)對象obj，這個(gè)時(shí)候obj的引用計(jì)數(shù)為2；A打算撤銷對obj的引用，完成第一步時(shí)引用計(jì)數(shù)減去1時(shí)，這時(shí)發(fā)生了線程切換，A掛起等待，還沒執(zhí)行銷毀對象操作。B進(jìn)入運(yùn)行狀態(tài)，這個(gè)時(shí)候B也對obj撤銷引用，并完成引用計(jì)數(shù)減1，銷毀對象，這個(gè)時(shí)候obj的引用數(shù)為0，釋放內(nèi)存。如果此時(shí)A重新喚醒，要繼續(xù)銷毀對象，可是這個(gè)時(shí)候已經(jīng)沒有對象了。所以為了保證不出現(xiàn)數(shù)據(jù)污染，才引入GIL。
每個(gè)線程使用前都會(huì)去獲取GIL權(quán)限，使用完釋放GIL權(quán)限。釋放線程的時(shí)機(jī)由python的另一個(gè)機(jī)制check_interval來決定。
在多核cpu時(shí)，因?yàn)樾枰@取和釋放GIL鎖，會(huì)存在性能上額外的損耗。特別是由于調(diào)度控制的原因，比如一個(gè)線程釋放了鎖，調(diào)度接著又分配cpu資源給同一個(gè)線程，該線程發(fā)起申請時(shí)，又重新獲得GIL，而其他線程實(shí)際上都在等待，白白浪費(fèi)了申請和釋放鎖的操作耗時(shí)。
python中的線程比較適合I/O密集型的操作（磁盤IO或者網(wǎng)絡(luò)IO）。
- 線程的使用
- import os
- import time
- import sys
- from concurrent import futures
- def to_do(info):
- for i in range(100000000):
- pass
- return info[0]
- MAX_WORKERS = 10
- param_list = []
- for i in range(5):
- param_list.append(('text%s' % i, 'info%s' % i))
- workers = min(MAX_WORKERS, len(param_list))
- # with 默認(rèn)會(huì)等所有任務(wù)都完成才返回，所以這里會(huì)阻塞
- with futures.ThreadPoolExecutor(workers) as executor:
- results = executor.map(to_do, sorted(param_list))
- # 打印所有
- for result in results:
- print(result)
- # 非阻塞的方式，適合不需要返回結(jié)果的情況
- workers = min(MAX_WORKERS, len(param_list))
- executor = futures.ThreadPoolExecutor(workers)
- results = []
- for idx, param in enumerate(param_list):
- result = executor.submit(to_do, param)
- results.append(result)
- print('result %s' % idx)
- # 手動(dòng)等待所有任務(wù)完成
- executor.shutdown()
- print('='*10)
- for result in results:
- print(result.result())
3. python中的進(jìn)程
python提供的multiprocessing包來規(guī)避GIL的缺點(diǎn)，實(shí)現(xiàn)在多核cpu上并行的目的。multiprocessing還提供進(jìn)程之間數(shù)據(jù)和內(nèi)存共享的機(jī)制。這里介紹的concurrent.futures的實(shí)現(xiàn)。用法和線程基本一樣，ThreadPoolExecutor改成ProcessPoolExecutor
- import os
- import time
- import sys
- from concurrent import futures
- def to_do(info):
- for i in range(10000000):
- pass
- return info[0]
- start_time = time.time()
- MAX_WORKERS = 10
- param_list = []
- for i in range(5):
- param_list.append(('text%s' % i, 'info%s' % i))
- workers = min(MAX_WORKERS, len(param_list))
- # with 默認(rèn)會(huì)等所有任務(wù)都完成才返回，所以這里會(huì)阻塞
- with futures.ProcessPoolExecutor(workers) as executor:
- results = executor.map(to_do, sorted(param_list))
- # 打印所有
- for result in results:
- print(result)
- print(time.time()-start_time)
- # 耗時(shí)0.3704512119293213s， 而線程版本需要14.935384511947632s
4. python中的協(xié)程
4.1 簡單協(xié)程
我們先來看下python是怎么實(shí)現(xiàn)協(xié)程的。答案是yield。以下例子的功能是實(shí)現(xiàn)計(jì)算移動(dòng)平均數(shù)
- from collections import namedtuple
- Result = namedtuple('Result', 'count average')
- # 協(xié)程函數(shù)
- def averager():
- total = 0.0
- count = 0
- average = None
- while True:
- term = yield None # 暫停，等待主程序傳入數(shù)據(jù)喚醒
- if term is None:
- break # 決定是否退出
- total += term
- count += 1
- average = total/count # 累計(jì)狀態(tài)，包括上一次的狀態(tài)
- return Result(count, average)
- # 協(xié)程的觸發(fā)
- coro_avg = averager()
- # 預(yù)激活協(xié)程
- next(coro_avg)
- # 調(diào)用者給協(xié)程提供數(shù)據(jù)
- coro_avg.send(10)
- coro_avg.send(30)
- coro_avg.send(6.5)
- try:
- coro_avg.send(None)
- except StopIteration as exc: # 執(zhí)行完成，會(huì)拋出StopIteration異常，返回值包含在異常的屬性value里
- result = exc.value
- print(result)
yield關(guān)鍵字有兩個(gè)含義：產(chǎn)出和讓步; 把yield的右邊的值產(chǎn)出給調(diào)用方，同時(shí)做出讓步，暫停執(zhí)行，讓程序繼續(xù)執(zhí)行。
上面的例子可知
- 協(xié)程用yield來控制流程，接收和產(chǎn)出數(shù)據(jù)
- next()：預(yù)激活協(xié)程
- send：協(xié)程從調(diào)用方接收數(shù)據(jù)
- StopIteration：控制協(xié)程結(jié)束， 同時(shí)獲取返回值
我們來回顧下1.3中協(xié)程的概念：本質(zhì)是為了能讓多組過程能不獨(dú)自占用完所有資源，在一個(gè)線程內(nèi)交叉執(zhí)行，達(dá)到高并發(fā)的目的。。上面的例子怎么解釋呢？
- 可以把一個(gè)協(xié)程單次一個(gè)任務(wù)，即移動(dòng)平均
- 每個(gè)任務(wù)可以拆分成小步驟（也可以說是子程序), 即每次算一個(gè)數(shù)的平均
- 如果多個(gè)任務(wù)需要執(zhí)行呢？怎么調(diào)用控制器在調(diào)用方
- 如果有10個(gè)，可以想象，調(diào)用在控制的時(shí)候隨機(jī)的給每個(gè)任務(wù)send的一個(gè)數(shù)據(jù)化，就會(huì)是多個(gè)任務(wù)在交叉執(zhí)行，達(dá)到并發(fā)的目的。
4.2 asyncio協(xié)程應(yīng)用包
asyncio即異步I/O, 如在高并發(fā)（如百萬并發(fā)）網(wǎng)絡(luò)請求。異步I/O即你發(fā)起一個(gè)I/O操作不必等待執(zhí)行結(jié)束，可以做其他事情。asyncio底層是協(xié)程的方式來實(shí)現(xiàn)的。我們先來看一個(gè)例子，了解下asyncio的五臟六腑。
- import time
- import asyncio
- now = lambda : time.time()
- # async定義協(xié)程
- async def do_some_work(x):
- print("waiting:",x)
- # await掛起阻塞， 相當(dāng)于yield， 通常是耗時(shí)操作
- await asyncio.sleep(x)
- return "Done after {}s".format(x)
- # 回調(diào)函數(shù)，和yield產(chǎn)出類似功能
- def callback(future):
- print("callback:",future.result())
- start = now()
- tasks = []
- for i in range(1, 4):
- # 定義多個(gè)協(xié)程，同時(shí)預(yù)激活
- coroutine = do_some_work(i)
- task = asyncio.ensure_future(coroutine)
- task.add_done_callback(callback)
- tasks.append(task)
- # 定一個(gè)循環(huán)事件列表，把任務(wù)協(xié)程放在里面，
- loop = asyncio.get_event_loop()
- try:
- # 異步執(zhí)行協(xié)程，直到所有操作都完成， 也可以通過asyncio.gather來收集多個(gè)任務(wù)
- loop.run_until_complete(asyncio.wait(tasks))
- for task in tasks:
- print("Task ret:",task.result())
- except KeyboardInterrupt as e: # 協(xié)程任務(wù)的狀態(tài)控制
- print(asyncio.Task.all_tasks())
- for task in asyncio.Task.all_tasks():
- print(task.cancel())
- loop.stop()
- loop.run_forever()
- finally:
- loop.close()
- print("Time:", now()-start)
上面涉及到的幾個(gè)概念：
- event_loop 事件循環(huán)：程序開啟一個(gè)無限循環(huán)，把一些函數(shù)注冊到事件循環(huán)上，當(dāng)滿足事件發(fā)生的時(shí)候，調(diào)用相應(yīng)的協(xié)程函數(shù)
- coroutine 協(xié)程：協(xié)程對象，指一個(gè)使用async關(guān)鍵字定義的函數(shù)，它的調(diào)用不會(huì)立即執(zhí)行函數(shù)，而是會(huì)返回一個(gè)協(xié)程對象。協(xié)程對象需要注冊到事件循環(huán)，由事件循環(huán)調(diào)用。
- task任務(wù)：一個(gè)協(xié)程對象就是一個(gè)原生可以掛起的函數(shù)，任務(wù)則是對協(xié)程進(jìn)一步封裝，其中包含了任務(wù)的各種狀態(tài)
- future: 代表將來執(zhí)行或沒有執(zhí)行的任務(wù)的結(jié)果。它和task上沒有本質(zhì)上的區(qū)別
- async/await 關(guān)鍵字：python3.5用于定義協(xié)程的關(guān)鍵字，async定義一個(gè)協(xié)程，await用于掛起阻塞的異步調(diào)用接口。從上面可知，asyncio通過事件的方式幫我們實(shí)現(xiàn)了協(xié)程調(diào)用方的控制權(quán)處理，包括send給協(xié)程數(shù)據(jù)等。我們只要通過async定義協(xié)程，await定義阻塞，然后封裝成future的task，放入循環(huán)的事件列表中，就等著返回?cái)?shù)據(jù)。
再來看一個(gè)http下載的例子，比如你想下載5個(gè)不同的url（同樣的，你想接收外部的百萬的請求）
- import time
- import asyncio
- from aiohttp import ClientSession
- tasks = []
- url = "https://www.baidu.com/{}"
- async def hello(url):
- async with ClientSession() as session:
- async with session.get(url) as response:
- response = await response.read()
- # print(response)
- print('Hello World:%s' % time.time())
- if __name__ == '__main__':
- loop = asyncio.get_event_loop()
- for i in range(5):
- task = asyncio.ensure_future(hello(url.format(i)))
- tasks.append(task)
- loop.run_until_complete(asyncio.wait(tasks))
4.3 協(xié)程的應(yīng)用場景
- 支撐高并發(fā)I/O情況，如寫支撐高并發(fā)的服務(wù)端
- 代替線程，提供并發(fā)性能
- tornado和gevent都實(shí)現(xiàn)了類似功能， 之前文章提到Twisted也是
5. 總結(jié)
本文分享關(guān)于python協(xié)程的概念和asyncio包的初步使用情況，同時(shí)也介紹了基本的相關(guān)概念，如進(jìn)程、線程、并發(fā)、并行等。希望對你有幫助，歡迎交流（@mintel）。簡要總結(jié)如下：
- 并發(fā)和并行不一樣，并行是同時(shí)執(zhí)行多個(gè)任務(wù)， 并發(fā)是在極短時(shí)間內(nèi)處理多個(gè)任務(wù)
- 多核cpu，進(jìn)程是并行，python線程受制于GIL，不能并行，反而因?yàn)樯舷挛那袚Q更耗時(shí)，協(xié)程正好可以彌補(bǔ)
- 協(xié)程也不是并行，只是任務(wù)交替執(zhí)行任務(wù)，在存在阻塞I/O情況，能夠異步執(zhí)行，提高效率
- asyncio 異步I/O庫，可用于開發(fā)高并發(fā)應(yīng)用
當(dāng)前名稱：5分鐘完全掌握Python協(xié)程
地址分享：http://m.5511xx.com/article/dpsgohd.html
