說(shuō)明

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

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

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

用的對(duì)象。這兩個(gè)類(lèi)在內(nèi)部維護(hù)著一個(gè)工作線(xiàn)程或進(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方法作用是向線(xiàn)程池提交可回調(diào)的task，并返回一個(gè)回調(diào)實(shí)例。

example：

import time
from concurrent.futures import ThreadPoolExecutor

# 可回調(diào)的task
def pub_task(msg):
  time.sleep(3)
  return msg

# 創(chuàng)建一個(gè)線(xiàn)程池
pool = ThreadPoolExecutor(max_workers=3)

# 往線(xiàn)程池加入2個(gè)task
task1 = pool.submit(pub_task, 'a')
task2 = pool.submit(pub_task, 'b')

print(task1.done())    # False
time.sleep(4)
print(task2.done())    # True

print(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 ThreadPoolExecutor
import requests

URLS = ['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操作，訪(fǎng)問(wèn)url時(shí)（如下）在等待結(jié)果返回之前會(huì)產(chǎn)生阻塞，cpu不能做其他事情，而Future的引入幫助我們?cè)诘却倪@段時(shí)間可以完成其他的操作。

from concurrent.futures import ThreadPoolExecutor
from concurrent.futures import as_completed
import requests

URLS = ['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 asyncio
import time
start = 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é)程與線(xiàn)程

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

必須記住保留鎖，去保護(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àn)程之間同步操作，協(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)稱(chē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è)參考，也希望大家多多支持腳本之家。

最新評(píng)論