各语言的异步实现方式对比

Post Views: 18

本文系统梳理 Python 和 TypeScript 中的异步关键字与机制，通过真实场景帮助理解”什么时候该用异步”以及”两者有何本质区别”。

一、先搞清楚：异步解决什么问题

在深入语法之前，先回答一个根本问题：什么场景值得用异步？

典型异步场景

场景	说明	阻塞方式
网络请求（HTTP / API）	等待服务器响应	等待网络 I/O
数据库读写	等待磁盘 / 网络	等待数据库 I/O
文件读写	等待磁盘	等待磁盘 I/O
消息队列	等待消息到达	等待 broker I/O
定时任务 / 延迟执行	等待时间流逝	时间回调

核心原则：CPU 密集的任务（矩阵运算、视频解码）不适合用异步；I/O 密集的任务才是异步的用武之地——因为在等待 I/O 时，CPU 实际上在闲置。

二、TypeScript 的异步关键字体系

TypeScript（运行在 V8 / Node.js 上）的异步基于 JavaScript 的事件循环（Event Loop）。

2.1 核心关键字

async    → 声明一个异步函数，始终返回 Promise
await    → 暂停 async 函数执行，等待 Promise 解析
Promise  → 代表异步操作的最终结果（成功或失败）

2.2 三者如何配合

// 基础结构
async function fetchData() {
  // async 让函数返回 Promise
  // await 暂停当前函数，等 Promise 完成再继续
  const result = await fetch('/api/data');
  const json = await result.json();
  return json;
}

执行流程：

调用 fetchData()
  → 返回 Promise（不阻塞调用者）
  → 函数内部执行到 await fetch(...) 时，当前帧暂停
  → 事件循环去处理其他任务
  → fetch 完成，Promise 决议
  → 事件循环将后续代码重新入队
  → await 之后的代码继续执行

2.3 并发控制：Promise.all vs Promise.allSettled

// 并发：所有请求同时发出，等全部完成
const [users, orders] = await Promise.all([
  fetch('/api/users').then(r => r.json()),
  fetch('/api/orders').then(r => r.json()),
]);

// 容错：即使某个请求失败，也不影响其他结果
const results = await Promise.allSettled([
  fetch('/api/a'),
  fetch('/api/b'),
  fetch('/api/c'),
]);

2.4 完整项目示例：电商数据聚合服务

场景：首页需要同时加载用户信息、商品列表、推荐内容，三者互不依赖，希望并行请求。

// types.ts
interface User { id: number; name: string; }
interface Product { id: number; title: string; price: number; }
interface Recommendation { productId: number; score: number; }

interface HomePageData {
  user: User | null;
  products: Product[];
  recommendations: Recommendation[];
}

// service.ts
async function fetchUser(): Promise<User | null> {
  const res = await fetch('/api/user/profile');
  if (!res.ok) return null;
  return res.json();
}

async function fetchProducts(): Promise<Product[]> {
  const res = await fetch('/api/products?limit=20');
  return res.json();
}

async function fetchRecommendations(userId: number): Promise<Recommendation[]> {
  const res = await fetch(`/api/recommendations?userId=${userId}`);
  return res.json();
}

// main.ts
async function loadHomePage(): Promise<HomePageData> {
  // 先拿用户（推荐需要 userId）
  const user = await fetchUser();

  // 用户存在时，商品和推荐可以并行
  if (user) {
    const [products, recs] = await Promise.all([
      fetchProducts(),
      fetchRecommendations(user.id),
    ]);
    return { user, products, recommendations: recs };
  }

  // 未登录：只并行加载商品
  const [products] = await Promise.all([
    fetchProducts(),
  ]);
  return { user: null, products, recommendations: [] };
}

这个例子的关键点：

await 串行处理有依赖关系的步骤（先拿用户，再发推荐）
Promise.all 并行处理无依赖的步骤（商品和推荐同时发）
用 allSettled 替代 all 可以防止某个接口挂了导致整个页面白屏

三、Python 的异步关键字体系

Python 的异步基于 asyncio 库，从 Python 3.5 开始正式引入。

3.1 核心关键字

async    → 声明一个协程函数（coroutine），返回 coroutine 对象
await    → 暂停协程执行，等待另一个 coroutine 或 awaitable
asyncio  → Python 标准库，提供事件循环和并发原语

3.2 三者如何配合

import asyncio

# Python 中 async 函数定义时用 async，调用时必须用 await 或交给事件循环
async def fetch_data():
    # 模拟异步 I/O（真正的 asyncio 会用 aiohttp、asyncpg 等）
    await asyncio.sleep(1)  # 非阻塞地等待 1 秒
    return {"status": "done"}

# 必须通过事件循环来运行协程
result = asyncio.run(fetch_data())

与 TypeScript 的关键区别：

	TypeScript	Python
运行环境	浏览器 / V8 / Node.js（内置事件循环）	需要显式导入 `asyncio`，用 `asyncio.run()` 启动
调用 async 函数	直接调用返回 Promise，可立即用 await	直接调用返回 coroutine 对象，不自动运行
顶层 await	支持（ES2022）	不支持，必须放在 `async def` 内部

3.3 并发控制：asyncio.gather vs asyncio.Task

import asyncio

async def fetch_user():
    await asyncio.sleep(0.5)
    return {"id": 1, "name": "Alice"}

async def fetch_orders():
    await asyncio.sleep(0.3)
    return [{"order_id": 101, "total": 99.9}]

async def fetch_recommendations(user_id):
    await asyncio.sleep(0.8)
    return [{"product_id": 5, "score": 0.95}]

async def load_homepage():
    # gather：并发执行多个协程，返回结果列表
    user, orders = await asyncio.gather(
        fetch_user(),
        fetch_orders(),
    )

    # 依赖 user 的结果才能发起推荐请求
    recs = await fetch_recommendations(user["id"])
    return {"user": user, "orders": orders, "recs": recs}

# 运行
asyncio.run(load_homepage())

3.4 完整项目示例：异步爬虫 + 数据入库

场景：从多个 API 拉取数据，写入数据库，全部 I/O 操作异步完成。

# main.py
import asyncio
import aiohttp
import asyncpg

# 数据库连接池（async 版本）
DB_DSN = "postgres://user:pass@localhost:5432/ecommerce"

async def init_db():
    return await asyncpg.create_pool(DB_DSN)

async def fetch_session(session: aiohttp.ClientSession, url: str) -> dict:
    async with session.get(url) as resp:
        return await resp.json()

async def save_products(pool, products: list):
    async with pool.acquire() as conn:
        await conn.executemany(
            "INSERT INTO products (id, title, price) VALUES ($1, $2, $3)",
            [(p["id"], p["title"], p["price"]) for p in products],
        )

async def fetch_and_save(session: aiohttp.ClientSession, pool, url: str, table: str):
    data = await fetch_session(session, url)
    await save_products(pool, data)
    return f"[OK] {table}: {len(data)} records"

async def main():
    pool = await init_db()
    urls = [
        ("https://api.example.com/products", "products"),
        ("https://api.example.com/categories", "categories"),
        ("https://api.example.com/users", "users"),
    ]

    # 并发抓取，超时保护
    async with aiohttp.ClientSession() as session:
        tasks = [
            fetch_and_save(session, pool, url, name)
            for url, name in urls
        ]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    for r in results:
        print(r if isinstance(r, str) else f"[ERR] {r}")

    await pool.close()

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

这个例子的关键点：

aiohttp 和 asyncpg 都是异步原生库，不能混用同步版
asyncio.gather 同时发起多个 HTTP 请求，总耗时取决于最慢的那个，而不是所有耗时之和
return_exceptions=True 让一个请求失败不中断其他请求

四、Python vs TypeScript 的核心区别

4.1 事件循环的启动方式

// TypeScript：事件循环内置，async 函数一调用就进入队列
async function work() {
  await fetch('/api/data');
}
work(); // 直接调用即可，运行时自动管理事件循环

# Python：需要显式启动事件循环
async def work():
    await asyncio.sleep(1)

# Python 3.7+ 推荐写法
asyncio.run(work())  # 启动并运行事件循环，直到协程完成

4.2 顶层异步代码

// TypeScript：ES2022 支持顶层 await
const data = await fetch('/api/config').then(r => r.json());
console.log(data);

# Python：顶层不能用 await，必须包在 async def 里
# 错误的写法：
# data = await fetch(...)  # ❌ SyntaxError

# 正确的写法：
async def main():
    data = await fetch(...)
    print(data)

asyncio.run(main())

4.3 错误处理

// TypeScript：try/catch 直接包裹 await
try {
  const data = await fetch('/api/data').then(r => r.json());
} catch (err) {
  console.error('请求失败:', err);
}

# Python：同样用 try/except，但要注意协程未 await 时的错误
try:
    data = await fetch(...)
except Exception as e:
    print(f'请求失败: {e}')

4.4 回调地狱 vs 扁平结构（两者相同）

// 都是 async/await 解决
const user = await getUser();
const orders = await getOrders(user.id);

# 完全对称
user = await get_user()
orders = await get_orders(user["id"])

4.5 生态差异

维度	TypeScript / Node.js	Python
HTTP 客户端	`fetch`（内置）、`axios`、`node-fetch`	`aiohttp`、`httpx`（需安装）
数据库驱动	`pg`、`mongoose`、`prisma`	`asyncpg`、`databases`、`sqlalchemy.ext.asyncio`
Web 框架	`Express`、`Fastify`、`NestJS`	`FastAPI`、`Starlette`
ORM	`Prisma`、`TypeORM`	`SQLAlchemy 2.0 async`

五、实际项目中的选择建议

用 TypeScript 异步的场景

前端 React/Vue 应用：调用后端 API、处理流式数据
Node.js 后端：高并发 I/O 服务（API 网关、实时推送、微服务编排）
全栈项目：前后端共享类型定义，异步逻辑统一风格

用 Python 异步的场景

异步爬虫和数据采集
需要结合 AI/数据处理（Python 生态优势）的 I/O 密集型服务
用 FastAPI 构建高性能异步 Web 服务

两者混用的常见模式

前端（TypeScript async/await）
    ↓ HTTP 请求
后端（Python FastAPI + asyncio）
    ↓ 异步查询
数据库（asyncpg / SQLAlchemy async）

六、速查表

概念	TypeScript	Python
异步函数声明	`async function f(){}`	`async def f():`
等待结果	`await promise`	`await coroutine`
并发执行多个	`Promise.all([a, b])`	`asyncio.gather(a, b)`
容错并发	`Promise.allSettled([...])`	`asyncio.gather(..., return_exceptions=True)`
启动运行	运行时自动	`asyncio.run(main())`
延迟执行	`setTimeout(fn, ms)`	`await asyncio.sleep(seconds)`
取消操作	`AbortController`	`task.cancel()`

七、一个对比练习

同一需求：并发请求 3 个接口，任一失败则返回错误，最后把所有结果合并。

TypeScript 实现

async function mergeAll(): Promise<Record<string, unknown>> {
  const [users, products, orders] = await Promise.all([
    fetch('/api/users').then(r => r.json()),
    fetch('/api/products').then(r => r.json()),
    fetch('/api/orders').then(r => r.json()),
  ]);
  return { users, products, orders };
}

Python 实现

async def merge_all() -> dict:
    users, products, orders = await asyncio.gather(
        fetch_json('/api/users'),
        fetch_json('/api/products'),
        fetch_json('/api/orders'),
    )
    return {"users": users, "products": products, "orders": orders}

两者逻辑完全对称，区别只在语法和启动方式。掌握一个后，另一个上手极快。

适用于 Python 3.11+ 和 TypeScript 5.x