python并发2之使用asyncio处理并发

# sinner_thread.py
import threading
import itertools
import time
import sys
class Signal: # 这个类定义一个可变对象，用于从外部控制线程
 go = True

def spin(msg, signal): # 这个函数会在单独的线程中运行，signal 参数是前边定义的Signal类的实例
 write, flush = sys.stdout.write, sys.stdout.flush
 for char in itertools.cycle('|/-\\'): # itertools.cycle 函数从指定的序列中反复不断地生成元素
  status = char + ' ' + msg
  write(status)
  flush()
  write('\x08' * len(status)) # 使用退格符把光标移回行首
  time.sleep(.1) # 每 0.1 秒刷新一次
  if not signal.go: # 如果 go属性不是 True，退出循环
   break

 write(' ' * len(status) + '\x08' * len(status)) # 使用空格清除状态消息，把光标移回开头

def slow_function(): # 模拟耗时操作
 # 假装等待I/O一段时间
 time.sleep(3) # 调用sleep 会阻塞主线程，这么做事为了释放GIL，创建从属线程
 return 42

def supervisor(): # 这个函数设置从属线程，显示线程对象，运行耗时计算，最后杀死进程
 signal = Signal()
 spinner = threading.Thread(target=spin,
        args=('thinking!', signal))
 print('spinner object:', spinner) # 显示线程对象 输出 spinner object: <Thread(Thread-1, initial)>
 spinner.start() # 启动从属进程
 result = slow_function() # 运行slow_function 行数，阻塞主线程。同时丛书线程以动画形式旋转指针
 signal.go = False
 spinner.join() # 等待spinner 线程结束
 return result
def main():
 result = supervisor() 
 print('Answer', result)
if __name__ == '__main__':
 main()

# spinner_asyncio.py
# 通过协程以动画的形式显示文本式旋转指针
import asyncio
import itertools
import sys
@asyncio.coroutine # 打算交给asyncio 处理的协程要使用 @asyncio.coroutine 装饰
def spin(msg):
 write, flush = sys.stdout.write, sys.stdout.flush
 for char in itertools.cycle('|/-\\'): # itertools.cycle 函数从指定的序列中反复不断地生成元素
  status = char + ' ' + msg
  write(status)
  flush()
  write('\x08' * len(status)) # 使用退格符把光标移回行首
  try:
   yield from asyncio.sleep(0.1) # 使用 yield from asyncio.sleep(0.1) 代替 time.sleep(.1), 这样的休眠不会阻塞事件循环
  except asyncio.CancelledError: # 如果 spin 函数苏醒后抛出 asyncio.CancelledError 异常，其原因是发出了取消请求
   break

 write(' ' * len(status) + '\x08' * len(status)) # 使用空格清除状态消息，把光标移回开头

@asyncio.coroutine
def slow_function(): # 5 现在此函数是协程，使用休眠假装进行I/O 操作时，使用 yield from 继续执行事件循环
 # 假装等待I/O一段时间
 yield from asyncio.sleep(3) # 此表达式把控制权交给主循环，在休眠结束后回复这个协程
 return 42

@asyncio.coroutine
def supervisor(): #这个函数也是协程，因此可以使用 yield from 驱动 slow_function
 spinner = asyncio.async(spin('thinking!')) # asyncio.async() 函数排定协程的运行时间，使用一个 Task 对象包装spin 协程，并立即返回
 print('spinner object:', spinner) # Task 对象，输出类似 spinner object: <Task pending coro=<spin() running at spinner_asyncio.py:6>>
 # 驱动slow_function() 函数，结束后，获取返回值。同事事件循环继续运行，
 # 因为slow_function 函数最后使用yield from asyncio.sleep(3) 表达式把控制权交给主循环
 result = yield from slow_function()
 # Task 对象可以取消；取消后会在协程当前暂停的yield处抛出 asyncio.CancelledError 异常
 # 协程可以捕获这个异常，也可以延迟取消，甚至拒绝取消
 spinner.cancel()
 return result

def main():
 loop = asyncio.get_event_loop() # 获取事件循环引用
 # 驱动supervisor 协程，让它运行完毕；这个协程的返回值是这次调用的返回值
 result = loop.run_until_complete(supervisor())
 loop.close()
 print('Answer', result)
if __name__ == '__main__':
 main()

res = yield from foo() # foo 可以是协程函数，也可以是返回 Future 或 task 实例的普通函数

asyncio.async(coro_or_future, *, loop=None)

import asyncio
def run_sync(coro_or_future):
 loop = asyncio.get_event_loop()
 return loop.run_until_complete(coro_or_future)
a = run_sync(some_coroutine())

import asyncio
import aiohttp # 需要pip install aiohttp
from flags import save_flag, show, main, BASE_URL
@asyncio.coroutine # 我们知道，协程应该使用 asyncio.coroutine 装饰
def get_flag(cc):
 url = "{}/{cc}/{cc}.gif".format(BASE_URL, cc=cc.lower())
  # 阻塞的操作通过协程实现，客户代码通过yield from 把指责委托给协程，以便异步操作
 resp = yield from aiohttp.request('GET', url) 
 # 读取也是异步操作
 image = yield from resp.read()
 return image

@asyncio.coroutine
def download_one(cc): # 这个函数也必须是协程，因为用到了yield from
 image = yield from get_flag(cc) 
 show(cc)
 save_flag(image, cc.lower() + '.gif')
 return cc

def download_many(cc_list):
 loop = asyncio.get_event_loop() # 获取事件序号底层实现的引用
 to_do = [download_one(cc) for cc in sorted(cc_list)] # 调用download_one 获取各个国旗，构建一个生成器对象列表
 # 虽然函数名称是wait 但它不是阻塞型函数，wait 是一个协程，等传给他的所有协程运行完毕后结束
 wait_coro = asyncio.wait(to_do)
 res, _ = loop.run_until_complete(wait_coro) # 执行事件循环，知道wait_coro 运行结束；事件循环运行的过程中，这个脚本会在这里阻塞。
 loop.close() # 关闭事件循环
 return len(res)
if __name__ == '__main__':
 main(download_many)

@asyncio.coroutine
def get_flag(cc):
 url = "{}/{cc}/{cc}.gif".format(BASE_URL, cc=cc.lower())
 resp = yield from aiohttp.request('GET', url) 
 image = yield from resp.read()
 return image
# 把yield form 去掉
def get_flag(cc):
 url = "{}/{cc}/{cc}.gif".format(BASE_URL, cc=cc.lower())
 resp = aiohttp.request('GET', url) 
 image = resp.read()
 return image

# 现在是不是清晰多了

import asyncio
import collections
from collections import namedtuple
from enum import Enum
import aiohttp
from aiohttp import web
from flags import save_flag, show, main, BASE_URL
DEFAULT_CONCUR_REQ = 5
MAX_CONCUR_REQ = 1000
Result = namedtuple('Result', 'status data')
HTTPStatus = Enum('Status', 'ok not_found error')

# 自定义异常用于包装其他HTTP货网络异常，并获取country_code，以便报告错误
class FetchError(Exception):
 def __init__(self, country_code):
  self.country_code = country_code

@asyncio.coroutine
def get_flag(cc):
 # 此协程有三种返回结果：
 # 1. 返回下载到的图片
 # 2. HTTP 响应为404 时，抛出web.HTTPNotFound 异常
 # 3. 返回其他HTTP状态码时， 抛出aiohttp.HttpProcessingError
 url = "{}/{cc}/{cc}.gif".format(BASE_URL, cc=cc.lower())
 resp = yield from aiohttp.request('GET', url)
 if resp.status == 200:
  image = yield from resp.read()
  return image
 elif resp.status == 404:
  raise web.HttpNotFound()
 else:
  raise aiohttp.HttpProcessionError(
   code=resp.status, message=resp.reason,
   headers=resp.headers
  )

@asyncio.coroutine
def download_one(cc, semaphore):
 # semaphore 参数是 asyncio.Semaphore 类的实例
 # Semaphore 类是同步装置，用于限制并发请求
 try:
  with (yield from semaphore):
    # 在yield from 表达式中把semaphore 当成上下文管理器使用，防止阻塞整个系统
    # 如果semaphore 计数器的值是所允许的最大值，只有这个协程会阻塞
    image = yield from get_flag(cc)
    # 退出with语句后 semaphore 计数器的值会递减，
    # 解除阻塞可能在等待同一个semaphore对象的其他协程实例
 except web.HTTPNotFound:
  status = HTTPStatus.not_found
  msg = 'not found'
 except Exception as exc:
  raise FetchError(cc) from exc
 else:
  save_flag(image, cc.lower() + '.gif')
  status = HTTPStatus.ok
  msg = 'ok'
 return Result(status, cc)

@asyncio.coroutine
def downloader_coro(cc_list):
 counter = collections.Counter()
 # 创建一个 asyncio.Semaphore 实例，最多允许激活MAX_CONCUR_REQ个使用这个计数器的协程
 semaphore = asyncio.Semaphore(MAX_CONCUR_REQ)
 # 多次调用 download_one 协程，创建一个协程对象列表
 to_do = [download_one(cc, semaphore) for cc in sorted(cc_list)]
 # 获取一个迭代器，这个迭代器会在future运行结束后返回future
 to_do_iter = asyncio.as_completed(to_do)
 for future in to_do_iter:
  # 迭代允许结束的 future 
  try:
   res = yield from future # 获取asyncio.Future 对象的结果（也可以调用future.result）
  except FetchError as exc:
   # 抛出的异常都包装在FetchError 对象里
   country_code = exc.country_code
   try:
    # 尝试从原来的异常 （__cause__）中获取错误消息
    error_msg = exc.__cause__.args[0]
   except IndexError:
    # 如果在原来的异常中找不到错误消息，使用所连接异常的类名作为错误消息
    error_msg = exc.__cause__.__class__.__name__
   if error_msg:
    msg = '*** Error for {}: {}'
    print(msg.format(country_code, error_msg))
   status = HTTPStatus.error
  else:
   status = res.status
  counter[status] += 1
 return counter

def download_many(cc_list):
 loop = asyncio.get_event_loop()
 coro = downloader_coro(cc_list)
 counts = loop.run_until_complete(coro)
 loop.close()
 return counts
if __name__ == '__main__':
 main(download_many)

with (yield from semaphore):
 image = yield from get_flag(cc)

@asyncio.coroutine
def download_one(cc, semaphore):
 try:
  with (yield from semaphore):
   image = yield from get_flag(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() # 获取事件循环的引用
  loop.run_in_executor(None, save_flag, image, cc.lower() + '.gif')
  status = HTTPStatus.ok
  msg = 'ok'
 return Result(status, cc)

def stage1(response1):
 request2 = step1(response1)
 api_call2(request2, stage2)
 
def stage2(response2):
 request3 = step3(response3)
 api_call3(request3, stage3) 

 def stage3(response3):
  step3(response3) 
api_call1(request1, stage1)

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

@asyncio.coroutine
def http_get(url):
 resp = yield from aiohttp.request('GET', url)
 if resp.status == 200:
  ctype = resp.headers.get('Content-type', '').lower()
  if 'json' in ctype or url.endswith('json'):
   data = yield from resp.json()
  else:
   data = yield from resp.read()
  return data
 elif resp.status == 404:
  raise web.HttpNotFound()
 else:
  raise aiohttp.HttpProcessionError(
   code=resp.status, message=resp.reason,
   headers=resp.headers)

@asyncio.coroutine
def get_country(cc):
 url = "{}/{cc}/metadata.json".format(BASE_URL, cc=cc.lower())
 metadata = yield from http_get(url)
 return metadata['country']

@asyncio.coroutine
def get_flag(cc):
 url = "{}/{cc}/{cc}.gif".format(BASE_URL, cc=cc.lower())
 return (yield from http_get(url))

@asyncio.coroutine
def download_one(cc, semaphore):
 try:
  with (yield from semaphore):
   image = yield from get_flag(cc)
  with (yield from semaphore):
   country = yield from get_country(cc)
 except web.HTTPNotFound:
  status = HTTPStatus.not_found
  msg = 'not found'
 except Exception as exc:
  raise FetchError(cc) from exc
 else:
  country = country.replace(' ', '_')
  filename = '{}--{}.gif'.format(country, cc)
  print(filename)
  loop = asyncio.get_event_loop()
  loop.run_in_executor(None, save_flag, image, filename)
  status = HTTPStatus.ok
  msg = 'ok'
 return Result(status, cc)

image = yield from http_get(url)
return image