Python asyncio并发编程详解：从GIL到async/await语法演进

阻塞型I/O和GIL

CPython解释器并非线程安全，它引入了全局解释器锁（GIL），确保同一时刻只有一个线程执行Python字节码。因此，单个Python进程通常无法利用多个CPU核心进行并行计算。

不过，标准库中所有执行阻塞型I/O操作的函数，在等待操作系统返回结果时会自动释放GIL。这意味着在Python语言层面可以使用多线程，且I/O密集型Python程序能从中受益：当一个线程等待网络响应时，阻塞型I/O函数会释放GIL，从而允许其他线程运行。

asyncio

该包使用由事件循环驱动的协程来实现并发。由于asyncio大量使用yield from表达式，它与Python旧版本不兼容。

asyncio包使用的“协程”定义较为严格。适用于asyncio API的协程在定义体中必须使用yield from，而不能使用普通的yield。此外，适合asyncio的协程需要由调用方驱动，并由调用方通过yield from进行调用。

示例1

import threading
import asyncio

@asyncio.coroutine
def hello():
  print('Start Hello', threading.currentThread())
  yield from asyncio.sleep(5)
  print('End Hello', threading.currentThread())

@asyncio.coroutine
def world():
  print('Start World', threading.currentThread())
  yield from asyncio.sleep(3)
  print('End World', threading.currentThread())

# 获取EventLoop:
loop = asyncio.get_event_loop()
tasks = [hello(), world()]
# 执行coroutine
loop.run_until_complete(asyncio.wait(tasks))
loop.close()

@asyncio.coroutine装饰器用于将生成器函数标记为协程类型。asyncio.sleep(3)创建一个3秒后完成的协程。loop.run_until_complete（future）用于运行直到future完成；如果参数是coroutine object，则需要使用ensure_future()函数进行包装。loop.close()用于关闭事件循环。

示例2

import asyncio

@asyncio.coroutine
def worker(text):
  """
  协程运行的函数
  :param text:
  :return:
  """
  i = 0
  while True:
    print(text, i)

    try:
      yield from asyncio.sleep(.1)
    except asyncio.CancelledError:
      break

    i += 1


@asyncio.coroutine
def client(text, io_used):
  worker_fu = asyncio.ensure_future(worker(text))

  # 假装等待I/O一段时间
  yield from asyncio.sleep(io_used)

  # 结束运行协程
  worker_fu.cancel()
  return 'done'


loop = asyncio.get_event_loop()
tasks = [client('xiaozhe', 3), client('zzzz', 5)]
result = loop.run_until_complete(asyncio.wait(tasks))
loop.close()
print('Answer:', result)

解释：

1. asyncio.ensure_future（coro_or_future, *, loop=None）：计划安排一个coroutine object的执行，返回一个asyncio.Task object。2. worker_fu.cancel()：取消一个协程的执行，抛出CancelledError异常。3. asyncio.wait()：协程的参数是一个由期物或协程构成的可迭代对象；wait会分别把各个协程包装进一个Task对象。

asyncio.Task对象与threading.Thread对象的比较

asyncio.Task对象大致与threading.Thread对象等效。Task对象用于驱动协程，而Thread对象用于调用可调用的对象。Task对象不由用户直接实例化，而是通过把协程传给asyncio.ensure_future(…)函数或loop.create_task(…)方法获取。获取的Task对象已经排定了运行时间；而Thread实例必须调用start方法，明确告知让它运行。如果想终止任务，可以使用Task.cancel()实例方法，这会在协程内部抛出CancelledError异常。

线程与协程的安全比较

如果使用线程进行过重要编程，由于调度程序随时可能中断线程，必须记住保留锁以保护程序中的重要部分，防止多步操作在执行过程中被中断，从而避免数据处于无效状态。

协程默认提供全方位保护，防止被意外中断。用户必须显式产出（yield）才能让程序的其余部分运行。对于协程而言，无需保留锁或在多个线程之间同步操作，因为协程自身就会同步，任意时刻只有一个协程运行。当想交出控制权时，可以使用yield或yield from将控制权交还调度程序。这就是能够安全地取消协程的原因：按照定义，协程只能在暂停的yield处取消，因此可以处理CancelledError异常并执行清理操作。

Future（期物）

通常情况下，用户不应自己创建期物，而只能由并发框架（concurrent.futures或asyncio）实例化。原因很简单：期物表示终将发生的事情，而确定某件事会发生的唯一方式是执行的时间已经排定。

asyncio.Future

在asyncio包中，BaseEventLoop.create_task(…)方法接收一个协程，排定它的运行时间，然后返回一个asyncio.Task实例——这也是asyncio.Future类的实例，因为Task是Future的子类，专门用于包装协程。

asyncio.ensure_future（coro_or_future, *, loop=None）

这个函数统一了协程和期物：第一个参数可以是二者中的任何一个。如果是Future或Task对象，则原封不动地返回。如果是协程，async函数会调用loop.create_task(…)方法创建Task对象。loop=关键字参数是可选的，用于传入事件循环；如果没有传入，async函数会通过调用asyncio.get_event_loop()函数获取循环对象。

BaseEventLoop.create_task(coro)

这个方法排定协程的执行时间，返回一个asyncio.Task对象。

asyncio包中有多个函数会自动把参数指定的协程包装在asyncio.Task对象中，例如BaseEventLoop.run_until_complete(…)方法。

asyncio.as_completed

为了集成进度条，我们可以使用as_completed生成器函数；幸好，asyncio包提供了这个生成器函数的相应版本。

使用asyncio和aiohttp包

从Python 3.4起，asyncio包只直接支持TCP和UDP。如果想使用HTTP或其他协议，则需要借助第三方包aiohttp。

cc_list = ['China', 'USA']

@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

@asyncio.coroutine
def download_one(name): 
  image = yield from get_flag(name) 
  save_flag(image, name.lower() + '.gif')
  return name

loop = asyncio.get_event_loop() 
wait_coro = asyncio.wait([download_one(cc) for cc in sorted(cc_list)]) 
res, _ = loop.run_until_complete(wait_coro) 
loop.close()

使用asyncio包时，我们编写的异步代码中包含由asyncio本身驱动的协程（即委派生成器），而生成器最终把职责委托给asyncio包或第三方库（如aiohttp）中的协程。这种处理方式相当于架起了管道，让asyncio事件循环（通过我们编写的协程）驱动执行低层异步I/O操作的库函数。

避免阻塞型调用

有两种方法能避免阻塞型调用中止整个应用程序的进程：1. 在单独的线程中运行各个阻塞型操作；2. 把每个阻塞型操作转换成非阻塞的异步调用使用。

多个线程是可以的，但是各个操作系统线程（Python使用的是这种线程）消耗的内存达兆字节（具体的量取决于操作系统种类）。如果要处理几千个连接，而每个连接都使用一个线程的话，我们负担不起。

把生成器当作协程使用是异步编程的另一种方式。对事件循环来说，调用回调与在暂停的协程上调用.send()方法效果差不多。各个暂停的协程是要消耗内存，但是比线程消耗的内存数量级小。

上面的脚本为什么会很快

在上面的脚本中，调用loop.run_until_complete方法时，事件循环驱动各个download_one协程，运行到第一个yield from表达式处时，那个表达式驱动各个get_flag协程，然后在get_flag协程里面运行到第一个yield from表达式处时，调用aiohttp.request(…)函数。这些调用都不会阻塞，因此在零点几秒内所有请求全部开始。

asyncio的基础设施获得第一个响应后，事件循环把响应发给等待结果的get_flag协程。得到响应后，get_flag向前执行到下一个yield from表达式处，调用resp.read()方法，然后把控制权还给主循环。其他响应会陆续返回。所有get_flag协程都获得结果后，委派生成器download_one恢复，保存图像文件。

async和await

为了简化并更好地标识异步IO，从Python 3.5开始引入了新的语法async和await，可以让coroutine的代码更简洁易读。

async和await是针对coroutine的新语法，要使用新的语法，只需要做两步简单的替换。1. 把@asyncio.coroutine替换为async；2. 把yield from替换为await。

例如：

@asyncio.coroutine
def hello():
  print("Hello world!")
  r = yield from asyncio.sleep(1)
  print("Hello again!")

等同于

async def hello():
  print("Hello world!")
  r = await asyncio.sleep(1)
  print("Hello again!")

网站请求实例

import asyncio
import aiohttp

urls = [
  'http://www.163.com/',
  'http://www.sina.com.cn/',
  'https://www.hupu.com/',
  'http://www.csdn.net/'
]


async def get_url_data(u):
  """
  读取url的数据
  :param u:
  :return:
  """
  print('running ', u)
  async with aiohttp.ClientSession() as session:
    async with session.get(u) as resp:
      print(u, resp.status, type(resp.text()))
      # print(await resp.text())

  return resp.headers


async def request_url(u):
  """
  主调度函数
  :param u:
  :return:
  """
  res = await get_url_data(u)
  return res


loop = asyncio.get_event_loop()
task_lists = asyncio.wait([request_url(u) for u in urls])
all_res, _ = loop.run_until_complete(task_lists)
loop.close()

print(all_res)

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持本站。