异步IO与事件驱动：asyncio核心概念、协程与任务

做量化交易这些年，我最大的感触就是——行情数据从来不等你。你想想看，一秒几百笔的Tick数据涌进来，如果用同步方式去处理，CPU大部分时间都在傻等IO。说白了，这就是浪费生命。

我个人习惯用asyncio来处理这类场景。它不是什么黑科技，就是一套让程序在等待IO时能干别的事的机制。今天我们就聊聊这个。

为什么需要异步IO？

先看一个真实场景。你在做行情推送，需要同时订阅多个交易所的深度数据。传统写法是这样的：

import time
import requests

def fetch_depth(exchange):
    resp = requests.get(f"https://{exchange}.com/api/depth")
    return resp.json()

# 串行请求，一个等一个
start = time.time()
for ex in ["binance", "okx", "bybit"]:
    data = fetch_depth(ex)
    print(f"{ex} depth received")
print(f"总耗时: {time.time() - start:.2f}s")

这段代码有什么问题？每次网络请求都在等。等DNS解析、等TCP握手、等响应体。CPU在这段时间里啥也没干，就干等着。

我在项目中遇到过类似情况，当时同时订阅了6个交易所的Level2数据，串行跑下来要3秒多。行情都变了两轮了，你拿到的还是旧数据。这显然不行。

asyncio的核心概念

asyncio的核心就三个东西：事件循环、协程、任务。我们一个个说。

事件循环

事件循环就是个调度器。它不断问：有没有IO事件完成了？有的话，叫醒对应的协程去处理。没有的话，继续等。

你可以把它想象成一个高效的管家。管家不会站在门口等快递，他会去做别的事。快递到了，门铃响了，他再去拿。

import asyncio

async def main():
    print("事件循环开始工作")
    await asyncio.sleep(1)
    print("事件循环还在跑")

# 启动事件循环
asyncio.run(main())

嗯，这里要注意：asyncio.run() 是Python 3.7引入的。老版本得用 loop.run_until_complete()。我建议新项目直接用 asyncio.run()，省心。

协程

协程就是加了 async def 的函数。它跟普通函数最大的区别是：可以暂停，可以恢复。

你想想看，普通函数一旦调用，必须从头跑到尾。协程不一样，遇到 await 就主动让出CPU，等条件满足了再回来继续跑。

async def fetch_data():
    print("开始获取数据...")
    # 模拟网络IO，主动让出CPU
    await asyncio.sleep(2)
    print("数据获取完成")
    return {"price": 45000}

这里 await asyncio.sleep(2) 不是真的在睡。它告诉事件循环：我要等2秒，你先去干别的。2秒后叫我。

任务

任务是对协程的进一步封装。它让协程可以并发执行。说白了，任务就是告诉事件循环：这个协程你帮我盯着，它暂停的时候你去跑别的。

async def handle_tick(symbol):
    print(f"处理 {symbol} 的Tick数据")
    await asyncio.sleep(0.1)  # 模拟处理耗时
    print(f"{symbol} 处理完成")

async def main():
    # 创建三个任务，并发执行
    tasks = [
        asyncio.create_task(handle_tick("BTC")),
        asyncio.create_task(handle_tick("ETH")),
        asyncio.create_task(handle_tick("SOL"))
    ]
    # 等待所有任务完成
    await asyncio.gather(*tasks)

asyncio.run(main())

这段代码里，三个 handle_tick 是同时跑的。总耗时约0.1秒，而不是0.3秒。这就是并发的威力。

在行情推送中的应用

好了，理论说完了。我们看看实际怎么用。下面是一个简化版的行情推送系统：

import asyncio
import random

class MarketDataFeed:
    """模拟行情推送"""
    
    def __init__(self, symbol):
        self.symbol = symbol
        self._running = False
    
    async def subscribe(self):
        """订阅行情"""
        self._running = True
        print(f"已订阅 {self.symbol} 行情")
        
        while self._running:
            # 模拟行情数据到达
            tick = {
                "symbol": self.symbol,
                "price": random.uniform(40000, 50000),
                "volume": random.randint(1, 100),
                "timestamp": asyncio.get_event_loop().time()
            }
            # 推送给处理器
            await self._process_tick(tick)
            # 模拟行情到达间隔
            await asyncio.sleep(0.01)  # 10ms一个Tick
    
    async def _process_tick(self, tick):
        """处理单笔Tick"""
        # 这里可以做策略计算、风控检查等
        # 为了演示，我们只打印
        print(f"{tick['symbol']}: {tick['price']:.2f} @ {tick['volume']}")
    
    def unsubscribe(self):
        self._running = False

async def main():
    # 同时订阅多个品种
    feeds = [
        MarketDataFeed("BTC/USDT"),
        MarketDataFeed("ETH/USDT"),
        MarketDataFeed("SOL/USDT")
    ]
    
    # 并发启动所有订阅
    tasks = [asyncio.create_task(feed.subscribe()) for feed in feeds]
    
    # 跑5秒后停止
    await asyncio.sleep(5)
    for feed in feeds:
        feed.unsubscribe()
    
    # 等待所有任务优雅退出
    await asyncio.gather(*tasks, return_exceptions=True)
    print("行情推送已停止")

if __name__ == "__main__":
    asyncio.run(main())

性能对比：同步 vs 异步

我做个简单测试，看看差距有多大。模拟10个品种，每个处理耗时50ms：

方式	代码	总耗时
同步	for循环串行	500ms
多线程	ThreadPoolExecutor	~55ms（有GIL开销）
异步	asyncio.gather	~52ms（纯IO等待）

异步的优势在IO密集型场景特别明显。如果是CPU密集型计算，那就得用多进程了。这个后面章节会讲。

常见陷阱与避坑指南

做异步编程，有几个坑我踩过，分享给你：

小结

asyncio不是什么魔法。它就是一套协作式多任务机制。协程主动让出CPU，事件循环调度其他任务，IO完成后恢复执行。

在量化交易里，行情推送、订单管理、风控检查这些场景，天生适合异步IO。你想想看，一秒几百笔Tick，每笔都要做策略计算、风控检查、日志记录。如果用同步方式，CPU大部分时间在等网络IO。用异步，这些等待时间就被利用起来了。

我个人建议，新写的行情处理系统，优先考虑asyncio。它比多线程更轻量，比多进程更容易调试。当然，具体选型要看场景。这个我们后面再细聊。

---