說(shuō)明

Python標(biāo)準(zhǔn)庫(kù)為我們提供了threading和multiprocessing模塊編寫(xiě)相應(yīng)的多線程/多進(jìn)程代碼。

從Python3.2開(kāi)始，標(biāo)準(zhǔn)庫(kù)為我們提供了concurrent.futures模塊，concurrent.futures 模塊的主要特色是 ThreadPoolExecutor 和

ProcessPoolExecutor 類，這兩個(gè)類實(shí)現(xiàn)的接口能分別在不同的線程或進(jìn)程中執(zhí)行可調(diào)

用的對(duì)象。這兩個(gè)類在內(nèi)部維護(hù)著一個(gè)工作線程或進(jìn)程池，以及要執(zhí)行的任務(wù)隊(duì)列。

Python 3.4 以后標(biāo)準(zhǔn)庫(kù)中asyncio 包，這個(gè)包使用事件循環(huán)驅(qū)動(dòng)的協(xié)程實(shí)現(xiàn)并發(fā)。這是 Python 中最大也

是最具雄心壯志的庫(kù)之一。asyncio 大量使用 yield from 表達(dá)式，因此與

Python 舊版不兼容。

submit和map方法

submit方法作用是向線程池提交可回調(diào)的task，并返回一個(gè)回調(diào)實(shí)例。

example：

import timefrom concurrent.futures import ThreadPoolExecutor# 可回調(diào)的taskdef pub_task(msg): time.sleep(3) return msg# 創(chuàng)建一個(gè)線程池pool = ThreadPoolExecutor(max_workers=3)# 往線程池加入2個(gè)tasktask1 = pool.submit(pub_task, ’a’)task2 = pool.submit(pub_task, ’b’)print(task1.done()) # Falsetime.sleep(4)print(task2.done()) # Trueprint(task1.result())print(task2.result())

map方法是創(chuàng)建一個(gè)迭代器，回調(diào)的結(jié)果有序放在迭代器中。

問(wèn)題：

Executor.map 函數(shù)易于使用，不過(guò)有個(gè)特性可能有用，也可能沒(méi)用，具體情況取決于需求：這個(gè)函數(shù)返回結(jié)果的順序與調(diào)用開(kāi)始的順序一致。

如果第一個(gè)調(diào)用生成結(jié)果用時(shí) 10秒，而其他調(diào)用只用 1 秒，代碼會(huì)阻塞 10 秒，獲取 map 方法返回的生成器產(chǎn)出的第一個(gè)結(jié)果。

在此之后，獲取后續(xù)結(jié)果時(shí)不會(huì)阻塞，因?yàn)楹罄m(xù)的調(diào)用已經(jīng)結(jié)束。

如果必須等到獲取所有結(jié)果后再處理，這種行為沒(méi)問(wèn)題；不過(guò)，通常更可取的方式是，不管提交的順序，只要有結(jié)果就獲取。

為此，要把 Executor.submit 方法和 futures.as_completed 函數(shù)結(jié)合起來(lái)使用。

from concurrent.futures import ThreadPoolExecutorimport requestsURLS = [’http://www.csdn.com’, ’http://qq.com’, ’http://www.leasonlove.cn’]def task(url, timeout=10): return requests.get(url, timeout=timeout)pool = ThreadPoolExecutor(max_workers=3)results = pool.map(task, URLS)for ret in results: print(’%s, %s’ % (ret.url, ret))

future異步編程

Future可以理解為一個(gè)在未來(lái)完成的操作，這是異步編程的基礎(chǔ)。通常情況下，我們執(zhí)行io操作，訪問(wèn)url時(shí)（如下）在等待結(jié)果返回之前會(huì)產(chǎn)生阻塞，cpu不能做其他事情，而Future的引入幫助我們?cè)诘却倪@段時(shí)間可以完成其他的操作。

from concurrent.futures import ThreadPoolExecutorfrom concurrent.futures import as_completedimport requestsURLS = [’http://www.csdn.cn’, ’http://qq.com’, ’http://www.leasonlove.cn’]def task(url, timeout=1): return requests.get(url, timeout=timeout)with ThreadPoolExecutor(max_workers=3) as executor: future_tasks = [executor.submit(task, url) for url in URLS] for f in future_tasks: if f.running(): print(’%s is running’ % str(f)) for f in as_completed(future_tasks): try: ret = f.done() if ret:f_ret = f.result()print(’%s, done, result: %s, %s’ % (str(f), f_ret.url, f_ret.content)) except Exception as e: # 第一個(gè)url無(wú)響應(yīng) f.cancel() print(str(e))

asyncio庫(kù)協(xié)程實(shí)現(xiàn)并發(fā)

對(duì)于gevent 和 asyncio 建議大家放棄Gevent，擁抱asyncio，asyncio是Python3.4以后標(biāo)準(zhǔn)庫(kù)。

而且由于Gevent直接修改標(biāo)準(zhǔn)庫(kù)里面大部分的阻塞式系統(tǒng)調(diào)用，包括socket、ssl、threading和 select等模塊，而變?yōu)閰f(xié)作式運(yùn)行。

但是我們無(wú)法保證你在復(fù)雜的生產(chǎn)環(huán)境中有哪些地方使用這些標(biāo)準(zhǔn)庫(kù)會(huì)由于打了補(bǔ)丁而出現(xiàn)奇怪的問(wèn)題。

import asyncioimport timestart = time.time()async def do(x): print(’Waiting: ’, x) await asyncio.sleep(x) return ’Finish after {}s’.format(x)task1 = do(1)task2 = do(2)task3 = do(4)tasks = [ asyncio.ensure_future(task1), asyncio.ensure_future(task2), asyncio.ensure_future(task3)]loop = asyncio.get_event_loop()loop.run_until_complete(asyncio.wait(tasks))for task in tasks: print(’Task result: ’, task.result())end = time.time()print(’TIME: ’, end - start)

協(xié)程與線程

如果使用線程做過(guò)重要的編程，你就知道寫(xiě)出程序有多么困難，因?yàn)檎{(diào)度程序任何時(shí)候都能中斷線程。

必須記住保留鎖，去保護(hù)程序中的重要部分，防止多步操作在執(zhí)行的過(guò)程中中斷，防止數(shù)據(jù)處于無(wú)效狀態(tài)。

而協(xié)程默認(rèn)會(huì)做好全方位保護(hù)，以防止中斷。我們必須顯式產(chǎn)出才能讓程序的余下部分運(yùn)行。

對(duì)協(xié)程來(lái)說(shuō)，無(wú)需保留鎖，在多個(gè)線程之間同步操作，協(xié)程自身就會(huì)同步，因?yàn)樵谌我鈺r(shí)刻只有一個(gè)協(xié)程運(yùn)行。

想交出控制權(quán)時(shí)，可以使用 yield 或 yield from 把控制權(quán)交還調(diào)度程序。

這就是能夠安全地取消協(xié)程的原因：按照定義，協(xié)程只能在暫停的 yield處取消，因此可以處理 CancelledError 異常，執(zhí)行清理操作。

補(bǔ)充知識(shí)：Python-什么時(shí)候使用yield？

簡(jiǎn)介

很多時(shí)候在python代碼中見(jiàn)到了yield，沒(méi)有系統(tǒng)學(xué)習(xí)過(guò)，自己也沒(méi)有用過(guò)。

yield語(yǔ)句延遲了語(yǔ)句的執(zhí)行，然后發(fā)送了一個(gè)值給調(diào)用者，但保留了一定的狀態(tài)去保證函數(shù)離開(kāi)之后可以繼續(xù)。當(dāng)繼續(xù)的時(shí)候，函數(shù)繼續(xù)執(zhí)行上一個(gè)的運(yùn)行狀態(tài)。這使得它的代碼可以隨著時(shí)間產(chǎn)生一系列的值，而不是立即執(zhí)行，然后像一個(gè)list一樣發(fā)送他們回來(lái)。

例子

例子1：

# A Simple Python program to demonstrate working # of yield # A generator function that yields 1 for first time, # 2 second time and 3 third time def simpleGeneratorFun(): yield 1 yield 2 yield 3 # Driver code to check above generator function for value in simpleGeneratorFun(): print(value)

返回語(yǔ)句發(fā)送一個(gè)特殊的值給它的調(diào)用者，而yield產(chǎn)生了一系列的值，當(dāng)我們想要遍歷一個(gè)序列的時(shí)候，我們應(yīng)該使用yield，但不想要把整個(gè)序列存儲(chǔ)在內(nèi)存中。

yield用于python的生成器（generator）。一個(gè)genertator 被定義得看起來(lái)像一個(gè)普通函數(shù)一樣，但它需要產(chǎn)生一個(gè)數(shù)字得時(shí)候，它使用yield，而不是使用return。如果一個(gè)函數(shù)里面定義了yield，那么它自動(dòng)稱為了一個(gè)generator函數(shù)。、

例子2：

# A Python program to generate squares from 1 # to 100 using yield and therefore generator # An infinite generator function that prints # next square number. It starts with 1 def nextSquare(): i = 1; # An Infinite loop to generate squares while True: yield i*i i += 1 # Next execution resumes # from this point # Driver code to test above generator # function for num in nextSquare(): if num > 100: break print(num)

輸出1，4，9…100

以上這篇Python并發(fā)concurrent.futures和asyncio實(shí)例就是小編分享給大家的全部?jī)?nèi)容了，希望能給大家一個(gè)參考，也希望大家多多支持好吧啦網(wǎng)。