asyncio

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

def test_task():
    a = async(asyncio.sleep(100)) # 协程不会在此处暂停，因为相当于启动了一个新线程
    print('-------------')
    yield from async(asyncio.sleep(1)) # 协程会在此处暂停
    yield from asyncio.sleep(1) # 协程会在此暂停，yield from Future() 也是同样效果
    print('+++++') # 在上一个 sleep 返回时，输出
    a.cancel() # 当前协程运行完毕，a 被销毁，对应的协程也会被销毁，所以整个程序不会运行 100s

loop = asyncio.get_event_loop()
loop.run_until_complete(test_task())

1
2
3
4
5
6

def download_many(cc_list):
    loop = asyncio.get_event_loop()  # 获取事件循环底层实现的引用
    to_do = [download_one(cc) for cc in sorted(cc_list)]  # <9>
    wait_coro = asyncio.wait(to_do)  # <10>
    res,  _ = loop.run_until_complete(wait_coro) # 执行事件循环, 直到 wait_coro 运行结束
    loop.close() # 关闭事件循环

1
2
3
4

#  在 yield from 表达式中把 semaphore 当成上下文管理器使用
# 这段代码保证, 任何时候都不会有超过 concur_req 个 get_flag 协程启动
with (yield from semaphore):
    image = yield from get_flag(base_url,  cc)

1
2

loop = asyncio.get_event_loop()
loop.run_in_executor(None,  save_flag,  image,  cc.lower() + '.gif')

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

import asyncio
import collections
import contextlib

import aiohttp
from aiohttp import web
import tqdm

from flags2_common import main, HTTPStatus, Result, save_flag

# default set low to avoid errors from remote site, such as
# 503 - Service Temporarily Unavailable
DEFAULT_CONCUR_REQ = 5
MAX_CONCUR_REQ = 1000


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


  # BEGIN FLAGS3_ASYNCIO
  @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()  # <1>
          else:
              data = yield from res.read()  # <2>
          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)  # <3>
      return metadata['country']


  @asyncio.coroutine
  def get_flag(base_url, cc):
      url = '{}/{cc}/{cc}.gif'.format(base_url, cc=cc.lower())
      # 必须在外层加上括号，直接写 return yield from， Python 解析器会报告句法错误
      return (yield from http_get(url)) # <4>

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

# BEGIN FLAGS2_ASYNCIO_EXECUTOR
@asyncio.coroutine
def download_one(cc, base_url, semaphore, verbose):
    try:  
        # 如果semaphore 计数器的值是所允许的最大值，只有这个协程会阻塞
        # 使用 一个 with 和使用 两个 with 有区别么？？？？
        with (yield from semaphore): # <5>
            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 X from Y 句法链接原来的异常
        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)
# END FLAGS2_ASYNCIO_EXECUTOR

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

@asyncio.coroutine
def downloader_coro(cc_list, base_url, verbose, concur_req):
    counter = collections.Counter()
    # 最多允许激活 concur_req 个使用这个计数器的协程
    semaphore = asyncio.Semaphore(concur_req)
    to_do = [download_one(cc, base_url, semaphore, verbose)
             for cc in sorted(cc_list)]
    #  获取一个迭代器，这个迭代器会在期物运行结束后返回期物
    to_do_iter = asyncio.as_completed(to_do)
    if not verbose:
        to_do_iter = tqdm.tqdm(to_do_iter, total=len(cc_list))
    for future in to_do_iter:
        try:
            # 获取 asyncio.Future 对象的结果，
            # 最简单的方法是使用 yield from，而不是调用future.result() 方法
            res = yield from future  
        except FetchError as exc:
            # asyncio.as_completed 函数返回的期物与传给 as_completed 函数的期物可能不同
            # 因此通过 FetchError 包装网络异常，并关联相应的国家代码
            country_code = exc.country_code
            try:
                # 尝试从原来的异常（ __cause__）中获取错误消息
                error_msg = exc.__cause__.args[0]
            except IndexError:
                # 如果在原来的异常中找不到错误消息，使用所链接异常的类名作为错误消息
                error_msg = exc.__cause__.__class__.__name__
            if verbose and error_msg:
                msg = '*** Error for {}: {}'
                print(msg.format(country_code, error_msg))
            status = HTTPStatus.error
        else:
            status = res.status

        counter[status] += 1

    return counter

1
2
3
4
5
6

def download_many(cc_list, base_url, verbose, concur_req):
    loop = asyncio.get_event_loop()
    coro = downloader_coro(cc_list, base_url, verbose, concur_req)
    counts = loop.run_until_complete(coro)
    loop.close()
    return counts

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

def slow_function():   # 假设这是耗时的计算
    time.sleep(3)  # 调用 sleep 阻塞主线程, 一定要这么做, 以便释放 GIL, 创建从属线程
    return 42

def supervisor():
    done = threading.Event()
    spinner = threading.Thread(target=spin,
                               args=('thinking!',  done))
    print('spinner object: ',  spinner)
    spinner.start()  # Thread 实例必须调用 start 方法, 明确告知让它运行
    # 相当于耗时操作阻塞了主线程。同时, 从属线程以动画形式显示旋转指针
    result = slow_function()
    done.set() # 向子线程发送终止信号
    spinner.join() # 等待 spinner 线程结束
    return result

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

@asyncio.coroutine  # <1>  asyncio 处理的协程要使用 @asyncio.coroutine 装饰
def spin(msg):   # <2>      不是强制要求, 但是强烈建议这么做
    write,  flush = sys.stdout.write,  sys.stdout.flush
    for char in itertools.cycle('|/-\\'):
        status = char + ' ' + msg
        write(status)
        flush()
        write('\x08' * len(status))
        try:
            yield from asyncio.sleep(.1)  # <3> 休眠不会阻塞事件循环
        except asyncio.CancelledError:   # <4> 发出了取消请求而触发的异常
            break
    write(' ' * len(status) + '\x08' * len(status))

@asyncio.coroutine
def slow_function():   # <5> 假装进行 I/O 操作时, 使用 yield from 继续执行事件循环
    # pretend waiting a long time for I/O
    yield from asyncio.sleep(3)  # <6> 控制权交给主循环, 在休眠结束后恢复这个协程
    return 42


@asyncio.coroutine
def supervisor():   # <7>
    # 排定 spin 协程的运行时间, 使用一个 Task 对象包装 spin 协程, 并立即返回
    spinner = asyncio.async(spin('thinking!'))  # 获取的 Task 对象已经排定了运行时间
    print('spinner object: ',  spinner)  # <9>
    # 使用 yield from 驱动 slow_function 函数, 同时, 事件循环继续运行
    result = yield from slow_function() # <10>
    # Task 对象可以取消; 取消后会在协程当前暂停的 yield 处抛出 asyncio.CancelledError异常
    # 协程可以捕获这个异常, 也可以延迟取消, 甚至拒绝取消
    spinner.cancel()  # <11>
    return result


def main():
    loop = asyncio.get_event_loop()  # <12> 获取事件循环的引用
    # 驱动 supervisor 协程, 让它运行完毕
    result = loop.run_until_complete(supervisor())  # <13>
    loop.close()
    print('Answer: ',  result)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

#!/usr/bin/env python3

# BEGIN TCP_CHARFINDER_TOP
import sys
import asyncio

from charfinder import UnicodeNameIndex  # <1> 构建名称索引, 提供查询方法

CRLF = b'\r\n'
PROMPT = b'?> '

index = UnicodeNameIndex()  # <2>

@asyncio.coroutine
def handle_queries(reader,  writer):   # <3> 要传给 asyncio.start_server 的函数
    while True:   # <4> 两个参数是 asyncio.StreamReader 和 asyncio.StreamWriter 对象
        writer.write(PROMPT)  # 把数据写入缓冲, 通常不会阻塞, 因此不是协程, 只是普通的函数
        yield from writer.drain()  # 刷新 writer 缓冲; 执行真正的I/O操作, 所以是协程
        data = yield from reader.readline()  # <7> 协程, 返回一个 bytes 对象。
        try:
            query = data.decode().strip()
        except UnicodeDecodeError:   # <8>
            query = '\x00'
        client = writer.get_extra_info('peername')  # <9> 返回与套接字连接的远程地址
        print('Received from {}:  {!r}'.format(client,  query))  # <10>
        if query:
            if ord(query[: 1]) < 32:   # <11> 如果收到控制字符或者空字符, 退出循环
                break
            lines = list(index.find_description_strs(query)) # <12>
            if lines:
                writer.writelines(line.encode() + CRLF for line in lines) # <13>
            writer.write(index.status(query,  len(lines)).encode() + CRLF) # <14>

            yield from writer.drain()  # <15> 刷新输出缓冲
            print('Sent {} results'.format(len(lines)))  # <16>

    print('Close the client socket')  # <17>
    writer.close()  # <18> 关闭 StreamWriter 流
# END TCP_CHARFINDER_TOP

# BEGIN TCP_CHARFINDER_MAIN
def main(address='127.0.0.1',  port=2323):   # <1>
    port = int(port)
    loop = asyncio.get_event_loop()
    # 返回的协程对象是一个 asyncio.Server 实例, 即一个 TCP 套接字服务器
    server_coro = asyncio.start_server(handle_queries,  address,  port,
                                loop=loop) # <2>
    # 驱动 server_coro 协程, 启动服务器（ server）
    server = loop.run_until_complete(server_coro) # <3>

    host = server.sockets[0].getsockname()  # <4>
    print('Serving on {}. Hit CTRL-C to stop.'.format(host))  # <5>
    try:
        loop.run_forever()  # <6>  运行事件循环;  main 函数在这里阻塞
    except KeyboardInterrupt:   # CTRL+C pressed
        pass

    print('Server shutting down.')
    server.close()  # <7> 关闭服务器
    # server.wait_closed() 方法返回一个期物，调用 loop.run_until_complete 方法运行期物
    loop.run_until_complete(server.wait_closed())  # <8>  
    loop.close()  # <9> 终止事件循环


if __name__ == '__main__':
    main(*sys.argv[1: ])  # <10>
# END TCP_CHARFINDER_MAIN

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

index = UnicodeNameIndex()
with open(TEMPLATE_NAME) as tpl:
    template = tpl.read()
template = template.replace('{links}',  LINKS_HTML)

# BEGIN HTTP_CHARFINDER_HOME
def home(request):   # 一个路由处理函数, 参数是一个 aiohttp.web.Request 实例
    query = request.GET.get('query',  '').strip()  # 获取查询字符串, 去掉首尾的空白
    print('Query:  {!r}'.format(query))  # <3>
    if query:   # <4>
        descriptions = list(index.find_descriptions(query))
        res = '\n'.join(ROW_TPL.format(**descr._asdict())
                        for descr in descriptions)
        msg = index.status(query,  len(descriptions))
    else:
        descriptions = []
        res = ''
        msg = 'Enter words describing characters.'

    html = template.format(query=query,  result=res,   # <5>
                           message=msg)
    print('Sending {} results'.format(len(descriptions)))  # <6>
    return web.Response(content_type=CONTENT_TYPE,  text=html) # <7>
# END HTTP_CHARFINDER_HOME


# BEGIN HTTP_CHARFINDER_SETUP
@asyncio.coroutine
def init(loop,  address,  port):   # init 协程产出一个服务器, 交给事件循环驱动
    app = web.Application(loop=loop)  #  aiohttp.web.Application 类表示 Web 应用
    app.router.add_route('GET',  '/',  home)  # 通过路由把 URL 模式映射到处理函数上
    # 返回一个 aiohttp.web.RequestHandler 实例, 根据 app 对象设置的路由处理 HTTP 请求
    handler = app.make_handler()  
    # create_server 方法创建服务器, 以 handler 为协议处理程序
    server = yield from loop.create_server(handler,
                                           address,  port)  # <5>
    return server.sockets[0].getsockname()  # <6>

def main(address="127.0.0.1",  port=8888):
    port = int(port)
    loop = asyncio.get_event_loop()
    # 运行 init 函数, 启动服务器, 获取服务器的地址和端口
    host = loop.run_until_complete(init(loop,  address,  port))  # <7>
    print('Serving on {}. Hit CTRL-C to stop.'.format(host))
    try:
        loop.run_forever()  # <8> main 函数会在这里阻塞
    except KeyboardInterrupt:   # CTRL+C pressed
        pass
    print('Server shutting down.')
    loop.close()  # <9>