Question

使用协程做面向事件编程，需要下一番功夫才能掌握，因此最好知道，与经典的回调式编程相比，协程有哪些改进。这就是本节的话题。

只要对回调式面向事件编程有一定的经验，就知道“回调地狱”这个术语：如果一个操作需要依赖之前操作的结果，那就得嵌套回调。如果要连续做 3 次异步调用，那就需要嵌套 3 层回调。示例 18-10 是一个使用 JavaScript 编写的例子。

示例 18-10　JavaScript 中的回调地狱：嵌套匿名函数，也称为灾难金字塔

api_call1(request1, function (response1) {
  // 第一步
  var request2 = step1(response1);

  api_call2(request2, function (response2) {
    // 第二步
    var request3 = step2(response2);

    api_call3(request3, function (response3) {
      // 第三步
      step3(response3);
    });
  });
});

在示例 18-10 中，api_call1、api_call2 和 api_call3 是库函数，用于异步获取结果。例如，api_call1 从数据库中获取结果，api_call2 从 Web 服务器中获取结果。这 3 个函数都有回调。在 JavaScript 中，回调通常使用匿名函数实现（在下述 Python 示例中分别把这 3 个回调命名为 stage1、stage2 和 stage3）。这里的 step1、step2 和 step3 是应用程序中的常规函数，用于处理回调接收到的响应。

示例 18-11 展示 Python 中的回调地狱是什么样子。

示例 18-11　Python 中的回调地狱：链式回调

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, stage1)

虽然示例 18-11 中的代码与示例 18-10 的排布方式差异很大，但是作用却完全相同。前述 JavaScript 示例也能改写成这种排布方式（但是这段 Python 代码不能改写成 JavaScript 那种风格，因为 lambda 表达式句法上有限制）。

示例 18-10 和示例 18-11 组织代码的方式导致代码难以阅读，也更难编写：每个函数做一部分工作，设置下一个回调，然后返回，让事件循环继续运行。这样，所有本地的上下文都会丢失。执行下一个回调时（例如 stage2），就无法获取 request2 的值。如果需要那个值，那就必须依靠闭包，或者把它存储在外部数据结构中，以便在处理过程的不同阶段使用。

在这个问题上，协程能发挥很大的作用。在协程中，如果要连续执行 3 个异步操作，只需使用 yield3 次，让事件循环继续运行。准备好结果后，调用 .send() 方法，激活协程。对事件循环来说，这种做法与调用回调类似。但是对使用协程式异步 API 的用户来说，情况就大为不同了：3 次操作都在同一个函数定义体中，像是顺序代码，能在处理过程中使用局部变量保留整个任务的上下文。请看示例 18-12。

示例 18-12　使用协程和 yield from 结构做异步编程，无需使用回调

@asyncio.coroutine
def three_stages(request1):
  response1 = yield from api_call1(request1)
  # 第一步
  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))  # 必须显式调度执行

与前面的 JavaScript 和 Python 示例相比，示例 18-12 容易理解多了：操作的 3 个步骤依次写在同一个函数中。这样，后续处理便于使用前一步的结果；而且提供了上下文，能通过异常来报告错误。

假设在示例 18-11 中处理 api_call2(request2, stage2) 调用（stage1 函数最后一行）时抛出了 I/O 异常，这个异常无法在 stage1 函数中捕获，因为 api_call2 是异步调用，还未执行任何 I/O 操作就会立即返回。在基于回调的 API 中，这个问题的解决方法是为每个异步调用注册两个回调，一个用于处理操作成功时返回的结果，另一个用于处理错误。一旦涉及错误处理，回调地狱的危害程度就会迅速增大。

与此相比，在示例 18-12 中，那个三步操作的所有异步调用都在同一个函数中（three_stages），如果异步调用 api_call1、api_call2 和 api_call3 会抛出异常，那么可以把相应的 yield from 表达式放在 try/except 块中处理异常。

这么做比陷入回调地狱好多了，但是我不会把这种方式称为协程天堂，毕竟我们还要付出代价。我们不能使用常规的函数，必须使用协程，而且要习惯 yield from——这是第一个障碍。只要函数中有 yield from，函数就会变成协程，而协程不能直接调用，即不能像示例 18-11 中那样调用 api_call1(request1, stage1) 来启动回调链。我们必须使用事件循环显式排定协程的执行时间，或者在其他排定了执行时间的协程中使用 yield from 表达式把它激活。如果示例 18-12 没有最后一行（loop.create_task(three_stages(request1))），那么什么也不会发生。

下面举个例子来实践这个理论。

每次下载发起多次请求

假设保存每面国旗时，我们不仅想在文件名中使用国家代码，还想加上国家名称。那么，下载每面国旗时要发起两个请求：一个请求用于获取国旗，另一个请求用于获取图像所在目录里的 metadata.json 文件（记录着国家名称）。

在同一个任务中发起多个请求，这对线程版脚本来说很容易：只需接连发起两次请求，阻塞线程两次，把国家代码和国家名称保存在局部变量中，在保存文件时使用。如果想在异步脚本中使用回调做到这一点，你会闻到回调地狱中飘来的硫磺味道：国家代码和名称要放在闭包中传来传去，或者保存在某个地方，在保存文件时使用，这么做是因为各个回调在不同的局部上下文中运行。协程和 yield from 结构能缓解这个问题。解决方法虽然没有使用多个线程那么简单，但是比链式或嵌套回调易于管理。

示例 18-13 是使用 asyncio 包下载国旗脚本的第 3 版，这一次国旗的文件名中有国家名称。 flags2_asyncio.py 脚本（示例 18-7 和示例 18-8）中的 download_many 函数和 downloader_coro 协程没变，有变化的是下面的内容。

download_one

现在，这个协程使用 yield from 把职责委托给 get_flag 协程和新添的 get_country 协程。

get_flag

这个协程的大多数代码移到新添的 http_get 协程中了，以便也能在 get_country 协程中使用。

get_country

这个协程获取国家代码相应的 metadata.json 文件，从文件中读取国家名称。

http_get

从 Web 获取文件的通用代码。

示例 18-13　 flags3_asyncio.py：再定义几个协程，把职责委托出去，每次下载国旗时发起两次请求

@asyncio.coroutine
def http_get(url):
  res = yield from aiohttp.request('GET', url)
  if res.status == 200:
    ctype = res.headers.get('Content-type', '').lower()
    if 'json' in ctype or url.endswith('json'):
      data = yield from res.json()  ➊
    else:
      data = yield from res.read()  ➋
    return data

  elif res.status == 404:
    raise web.HTTPNotFound()
  else:
    raise aiohttp.errors.HttpProcessingError(
      code=res.status, message=res.reason,
      headers=res.headers)


@asyncio.coroutine
def get_country(base_url, 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(base_url, cc):
  url = '{}/{cc}/{cc}.gif'.format(base_url, cc=cc.lower())
  return (yield from http_get(url))  ➍


@asyncio.coroutine
def download_one(cc, base_url, semaphore, verbose):
  try:
    with (yield from semaphore):  ➎
      image = yield from get_flag(base_url, cc)
    with (yield from semaphore):
      country = yield from get_country(base_url, 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)
    loop = asyncio.get_event_loop()
    loop.run_in_executor(None, save_flag, image, filename)
    status = HTTPStatus.ok
    msg = 'OK'

  if verbose and msg:
    print(cc, msg)

  return Result(status, cc)

❶ 如果内容类型中包含 'json'，或者 url 以 .json 结尾，那么在响应上调用 .json() 方法，解析响应，返回一个 Python 数据结构——在这里是一个字典。

❷ 否则，使用 .read() 方法读取原始字节。

❸ metadata 变量的值是一个由 JSON 内容构建的 Python 字典。

❹ 这里必须在外层加上括号，如果直接写 return yield from，Python 解析器会不明所以，报告句法错误。

❺ 我分别在 semaphore 控制的两个 with 块中调用 get_flag 和 get_country，因为我想尽量缩减下载时间。

在示例 18-13 中，yield from 句法出现了 9 次。现在，你应该已经熟知如何在协程中使用这个结构把职责委托给另一个协程，而不阻塞事件循环。

问题的关键是，知道何时该使用 yield from，何时不该使用。基本原则很简单，yield from 只能用于协程和 asyncio.Future 实例（包括 Task 实例）。可是有些 API 很棘手，肆意混淆协程和普通的函数，例如下一节实现某个服务器时使用的 StreamWriter 类。

示例 18-13 是本书最后一次讨论 flags2 系列示例。我建议你自己运行那些示例，有助于对 HTTP 并发客户端的运作方式建立直观认识。你可以使用 -a、-e 和 -l 这三个命令行选项控制下载的国旗数量，还可以使用 -m 选项设置并发下载数。此外，还可以分别使用 LOCAL、REMOTE、DELAY 和 ERROR 服务器测试，找出能最大限度地利用各个服务器的吞吐量的并发下载数。如果想去掉错误或延迟，可以修改 vaurien_error_delay.sh 脚本中的设置。

客户端脚本到此结束，接下来使用 asyncio 包编写服务器。