用 asyncio 编写异步代码

等待却不停工

数据程序大部分时间并不在计算，而是在等待：等价格服务器回应、等订单被确认、等文件下载完。如果一个个地请求十个价格，就要连续等十次。异步代码让一个程序发出请求后把它先放在一边等待，同时去做别的事，而且全在一个线程里。Python 为此提供 asyncio，由 async def 和 await 两个关键字构成。

# A coroutine: calling it does nothing until it is awaited
import asyncio

async def get_price(symbol):
    await asyncio.sleep(0.01)                  # stands in for waiting on a network answer
    return {"AAPL": 184.25, "MSFT": 410.5}[symbol]

coro = get_price("AAPL")
print(type(coro).__name__)                     # coroutine  nothing has run yet
print(asyncio.run(coro))                       # 184.25     the event loop runs it

async def、await 与事件循环

用 async def 定义的函数是协程函数：调用它只返回一个协程对象，函数体一行也不会执行。在其中，await 某物 会让协程暂停，直到某物就绪，并把控制权交还给事件循环，循环在这期间可以运行其他协程。asyncio.run(coro) 启动一个循环并把一个协程运行到结束。await 只能出现在 async def 里；忘记它是经典错误，因为调用得到的是协程而不是答案。

# Awaiting ten prices one after another waits ten times; gather overlaps the waits
import asyncio
import time

async def get_price(symbol):
    await asyncio.sleep(0.05)
    return len(symbol) * 10.0

async def one_by_one(symbols):
    return [await get_price(s) for s in symbols]

async def together(symbols):
    return await asyncio.gather(*(get_price(s) for s in symbols))

symbols = ["AAPL", "MSFT", "NVDA", "TSLA", "AMZN", "META", "GOOG", "NFLX", "ORCL", "INTC"]
t0 = time.perf_counter()
asyncio.run(one_by_one(symbols))
slow = time.perf_counter() - t0
t0 = time.perf_counter()
asyncio.run(together(symbols))
fast = time.perf_counter() - t0
print(f"one by one: {slow:.2f} s, gathered: {fast:.2f} s")      # about 0.50 s against 0.05 s
print(slow > 4 * fast)                                           # True

错误、超时与手动驱动

等待可能失败。永远不回应的请求应该用 asyncio.wait_for(coro, timeout) 截断，它会抛出 TimeoutError；try / except 里的 await 处理错误的方式与普通代码完全一样，所以重试就是围着一个被等待的调用的循环。在底层，协程就是可以一步步推进的东西：coro.send(None) 运行到下一次暂停，结束时抛出带着结果的 StopIteration。事件循环就是同时对许多协程不断做这件事的代码，本实验的测试也是这样运行你的代码的。

# Stepping a coroutine by hand: this is what an event loop does for many of them
class Pause:
    def __await__(self):
        yield

async def job():
    await Pause()
    await Pause()
    return 184.25

coro = job()
steps = 0
while True:
    try:
        coro.send(None)
    except StopIteration as stop:
        print("finished with", stop.value)                  # finished with 184.25
        break
    steps += 1
print("paused", steps, "times")                              # paused 2 times

# A timeout turns a slow answer into an error you can handle, then retry
import asyncio

async def slow_feed():
    await asyncio.sleep(1.0)
    return 184.25

async def main():
    for attempt in range(1, 4):
        try:
            return await asyncio.wait_for(slow_feed(), timeout=0.01)
        except TimeoutError:
            print(f"attempt {attempt} timed out")
    return None

print(asyncio.run(main()))                                   # three timeouts, then None

• async def 定义协程函数；调用它只创建协程，还没有任何东西运行。
• await 暂停到结果就绪，并让事件循环运行其他协程；只能用在 async def 内。
• asyncio.run 启动循环，gather 让等待重叠，wait_for 增加超时；错误用普通的 try / except 处理。

下一节：用 websocket 接收流式数据，实时接收行情 tick。

实验：评分内容

• 用 get_price 等待一个报价25 分
• 用 get_prices 获取自选列表25 分
• 用 retry_price 应对超时25 分
• 用 drive 扮演事件循环25 分