Python中的并發(fā)處理之asyncio包使用的詳解

import asyncio
import itertools
import sys

@asyncio.coroutine # 交給 asyncio 處理的協(xié)程要使用 @asyncio.coroutine 裝飾
def spin(msg):
  for char in itertools.cycle('|/-\\'):
    status = char + ' ' + msg
    print(status)
    try:
      yield from asyncio.sleep(.1) # 使用 yield from asyncio.sleep(.1) 代替 time.sleep(.1)，這樣的休眠不會阻塞事件循環(huán)。
    except asyncio.CancelledError: # 如果 spin 函數(shù)蘇醒后拋出 asyncio.CancelledError 異常，其原因是發(fā)出了取消請求，因此退出循環(huán)。
      break

@asyncio.coroutine
def slow_function(): # slow_function 函數(shù)是協(xié)程，在用休眠假裝進行 I/O 操作時，使用 yield from 繼續(xù)執(zhí)行事件循環(huán)。
  # 假裝等待I/O一段時間
  yield from asyncio.sleep(3) # yield from asyncio.sleep(3) 表達式把控制權交給主循環(huán)，在休眠結束后恢復這個協(xié)程。
  return 42

@asyncio.coroutine
def supervisor(): # supervisor 函數(shù)也是協(xié)程
  spinner = asyncio.async(spin('thinking!')) # asyncio.async(...) 函數(shù)排定 spin 協(xié)程的運行時間，使用一個 Task 對象包裝spin 協(xié)程，并立即返回。
  print('spinner object:', spinner)
  result = yield from slow_function() # 驅動 slow_function() 函數(shù)。結束后，獲取返回值。
# 同時，事件循環(huán)繼續(xù)運行，因為slow_function 函數(shù)最后使用 yield from asyncio.sleep(3) 表達式把控制權交回給了主循環(huán)。
  spinner.cancel() # Task 對象可以取消；取消后會在協(xié)程當前暫停的 yield 處拋出 asyncio.CancelledError 異常。協(xié)程可以捕獲這個異常，也可以延遲取消，甚至拒絕取消。
  return result

if __name__ == '__main__':
  loop = asyncio.get_event_loop() # 獲取事件循環(huán)的引用
  result = loop.run_until_complete(supervisor()) # 驅動 supervisor 協(xié)程，讓它運行完畢；這個協(xié)程的返回值是這次調用的返回值。
  loop.close()
  print('Answer:', result)

@asyncio.coroutine
def download_one(cc, base_url, semaphore, verbose):
  try:
    with (yield from semaphore):
      image = yield from get_flag(base_url, cc)
  except web.HTTPNotFound:
    status = HTTPStatus.not_found
    msg = 'not found'
  except Exception as exc:
    raise FetchError(cc) from exc
  else:
    loop = asyncio.get_event_loop() # 獲取事件循環(huán)對象的引用
    loop.run_in_executor(None, # None 使用默認的 TrreadPoolExecutor 實例
        save_flag, image, cc.lower() + '.gif') # 傳入可調用對象
    status = HTTPStatus.ok
    msg = 'OK'

  if verbose and msg:
    print(cc, msg)

  return Result(status, cc)

def stage1(response1):
  request2 = step1(response1)
  api_call2(request2, stage2)

def stage2(response2):
  request3 = step2(response2)
  api_call3(request3, stage3)

def stage3(response3):
  step3(response3)

api_call1(request1, step1)

@asyncio.coroutine
def three_stages(request1):
  response1 = yield from api_call1()
  request2 = step1(response1)
  response2 = yield from api_call2(request2)
  request3 = step2(response2)
  response3 = yield from api_call3(request3)
  step3(response3)

loop.create_task(three_stages(request1))
# 協(xié)程不能直接調用，必須用事件循環(huán)顯示指定協(xié)程的執(zhí)行時間，或者在其他排定了執(zhí)行時間的協(xié)程中使用 yield from 表達式把它激活