### 1. 最简单的应用
```python
from fastapi import FastAPI

app = FastAPI()  # 创建应用实例

@app.get("/")  # 路由装饰器
async def root():
    return {"message": "Hello World"}
```

保存为 `main.py`，运行：

```bash
uv run uvicorn main:app --reload
```

访问 `http://127.0.0.1:8000`，你会看到：

```json
{"message": "Hello World"}
```

### 2. 路径参数

```python
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    """
    item_id 会自动转换为 int 类型
    访问: /items/123
    """
    return {"item_id": item_id}
```

### 3. 查询参数

```python
from typing import Optional

@app.get("/items/")
async def read_item(skip: int = 0, limit: int = 10):
    """
    查询参数: ?skip=0&limit=10
    """
    return {"skip": skip, "limit": limit}
```

---

**快速**: 与 NodeJS 和 Go 相当的高性能

| 特性 | FastAPI | Flask | Django |
|------|---------|-------|--------|
| 性能 | ⚡⚡⚡ | ⚡⚡ | ⚡ |
| 学习曲线 | 低 | 低 | 中 |
| 自动文档 | ✅ | ❌ | ❌ |
| 类型验证 | ✅ | ❌ | ⚠️ |
| 异步支持 | ✅ | ⚠️ | ✅ |

### 异步支持示例

FastAPI 原生支持异步编程，可以处理大量并发请求：

#### 1. 基础异步函数

```python
import asyncio
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    # 异步函数，不会阻塞其他请求
    await asyncio.sleep(1)  # 模拟IO操作
    return {"message": "异步响应"}
```

#### 2. 异步HTTP请求

```python
import httpx

@app.get("/weather/{city}")
async def get_weather(city: str):
    # 使用 httpx 异步客户端
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://api.weather.com/{city}"
        )
    return response.json()
```

#### 3. 并发处理多个请求

```python
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # 可以同时查询多个数据源
    user_data, user_posts = await asyncio.gather(
        fetch_user_from_db(user_id),
        fetch_user_posts(user_id)
    )
    return {"user": user_data, "posts": user_posts}

async def fetch_user_from_db(user_id: int):
    # 模拟数据库查询
    await asyncio.sleep(0.1)
    return {"id": user_id, "name": "张三"}

async def fetch_user_posts(user_id: int):
    # 模拟获取用户文章
    await asyncio.sleep(0.1)
    return [{"title": "文章1"}, {"title": "文章2"}]
```

**异步优势**：
- 在IO密集型操作中性能提升显著
- 可以同时处理数千个并发连接
- 不需要额外的线程或进程
- 适合数据库查询、API调用等场景

#### 4. 并发安全与资源竞争问题

**问题场景：多个请求修改同一资源**

```python
from fastapi import FastAPI, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession

# ❌ 危险：没有并发控制
@app.post("/wallet/add/{user_id}")
async def add_money(user_id: int, amount: int):
    # 两个请求同时执行：
    # 请求A: 读到balance=100
    # 请求B: 读到balance=100
    # 请求A: 写入150 (+50)
    # 请求B: 写入130 (+30)  ← A的修改丢失了！
    balance = await get_balance(user_id)  # balance=100
    new_balance = balance + amount
    await update_balance(user_id, new_balance)
    return {"balance": new_balance}
```

**解决方案1：使用数据库事务（推荐）**

```python
from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncSession

@app.post("/wallet/add/{user_id}")
async def add_money(
    user_id: int,
    amount: int,
    db: AsyncSession
):
    # 使用数据库行锁（FOR UPDATE）
    async with db.begin():  # 自动事务
        # SELECT ... FOR UPDATE - 锁定该行
        stmt = select(User).where(
            User.id == user_id
        ).with_for_update()
        result = await db.execute(stmt)
        user = result.scalar_one()

        # 其他请求会等待锁释放
        user.balance += amount
        await db.flush()  # 提交后自动释放锁

    return {"balance": user.balance}
```

**解决方案2：使用分布式锁（Redis）**

```python
import aioredis
from contextlib import asynccontextmanager

redis = aioredis.from_url("redis://localhost")

@asynccontextmanager
async def redis_lock(lock_key: str, expire: int = 10):
    """分布式锁上下文管理器"""
    lock = None
    try:
        lock = await redis.set(
            lock_key, "locked",
            nx=True, ex=expire
        )
        if not lock:
            raise HTTPException(
                400, "操作正在进行，请稍后重试"
            )
        yield
    finally:
        if lock:
            await redis.delete(lock_key)

@app.post("/wallet/add/{user_id}")
async def add_money(user_id: int, amount: int):
    lock_key = f"lock:user:{user_id}"

    # 获取锁，同一用户的请求串行执行
    async with redis_lock(lock_key):
        balance = await get_balance(user_id)
        new_balance = balance + amount
        await update_balance(user_id, new_balance)

    return {"balance": new_balance}
```

**解决方案3：原子操作（最佳性能）**

```python
@app.post("/wallet/add/{user_id}")
async def add_money(user_id: int, amount: int):
    # 直接使用SQL原子操作
    await db.execute(
        update(User)
        .where(User.id == user_id)
        .values(balance=User.balance + amount)  # 原子递增
    )
    return {"message": "success"}
```

**并发控制机制对比**：

| 方案 | 适用场景 | 性能 | 一致性 |
|------|---------|------|--------|
| 数据库事务 | 单数据库应用 | ⚡⚡⚡ | 强一致 |
| 分布式锁 | 多服务/缓存 | ⚡⚡ | 强一致 |
| 原子操作 | 简单递增/递减 | ⚡⚡⚡ | 强一致 |
| 乐观锁 | 读多写少 | ⚡⚡⚡ | 最终一致 |

**关键要点**：
- `await` 本身不保证线程安全
- 多个请求可以同时执行到同一个 `await` 之前
- 必须使用锁、事务等机制保护共享资源
- 优先使用数据库级别的原子操作

 

#### GET 请求

```python
@app.get("/users/")
async def get_users():
    return [{"id": 1, "name": "张三"}, {"id": 2, "name": "李四"}]
```

#### POST 请求

```python
from pydantic import BaseModel

class User(BaseModel):
    name: str
    age: int

@app.post("/users/")
async def create_user(user: User):
    """
    请求体:
    {
        "name": "张三",
        "age": 25
    }
    """
    return {"message": f"用户 {user.name} 创建成功", "user": user}
```
    

 
#### PUT 请求
```python
@app.put("/users/{user_id}")
async def update_user(user_id: int, user: User):
    return {"message": f"用户 {user_id} 更新成功"}
```

#### DELETE 请求
```python
@app.delete("/users/{user_id}")
async def delete_user(user_id: int):
    return {"message": f"用户 {user_id} 删除成功"}
```

 
- await 本身不保证线程安全
- 多个请求可以同时执行到同一个 await 之前 
  - await只是让等待一个异步线程执行完成，却不阻止其他线程访问该方法 

- 必须使用锁、事务等机制保护共享资源
  - 优先使用数据库级别的原子操作
    

 
在 FastAPI 中，`await` 和非同步编程涉及到一些重要概念。让我详细解释一下：

## 1. **异步模型基础**

在 FastAPI（基于 asyncio）中：
- **没有真正的多线程并发执行**：asyncio 使用单线程事件循环
- `await` 不会阻塞线程，而是让出控制权给事件循环
- 看似“同时”执行，实际是协程间的协作式多任务

## 2. **代码示例演示**

```python
from fastapi import FastAPI
import asyncio
import time

app = FastAPI()

# 共享资源示例
counter = 0
shared_data = {}

@app.get("/async-endpoint1")
async def endpoint1():
    global counter
    # 模拟异步IO操作
    await asyncio.sleep(1)
    current = counter
    await asyncio.sleep(0.1)  # 模拟处理时间
    counter = current + 1
    return {"endpoint": 1, "counter": counter}

@app.get("/async-endpoint2")
async def endpoint2():
    global counter
    await asyncio.sleep(0.5)
    current = counter
    await asyncio.sleep(0.2)
    counter = current + 1
    return {"endpoint": 2, "counter": counter}
```

## 3. **并发访问机制**

### 场景分析：
```python
# 假设两个请求几乎同时到达：
# 请求1: endpoint1()
# 请求2: endpoint2()

# 执行时序可能如下：
# 时间点 0ms: endpoint1() 开始，读取 counter=0
# 时间点 0ms: endpoint2() 开始，读取 counter=0
# 时间点 100ms: endpoint1() 设置 counter=1
# 时间点 200ms: endpoint2() 设置 counter=1（覆盖了之前的修改）
# 结果：counter=1，但预期应该是2
```

## 4. **数据库并发问题**

对于数据库同一行数据的更新：

### **乐观锁机制**：
```python
# 使用版本号或时间戳
# SQL示例：
"""
UPDATE users 
SET balance = balance + 100, version = version + 1
WHERE id = 1 AND version = {current_version}
"""
# 检查 affected_rows，如果是0，说明版本已过期，需要重试
```

### **悲观锁机制**：
```python
# 在事务中使用 SELECT ... FOR UPDATE
from databases import Database

async def transfer_funds(db: Database, from_id: int, to_id: int, amount: float):
    async with db.transaction():
        # 锁定两行数据
        query1 = "SELECT balance FROM accounts WHERE id = :id FOR UPDATE"
        from_balance = await db.fetch_val(query1, values={"id": from_id})
        
        if from_balance >= amount:
            # 执行更新
            update1 = "UPDATE accounts SET balance = balance - :amount WHERE id = :id"
            await db.execute(update1, values={"id": from_id, "amount": amount})
            
            update2 = "UPDATE accounts SET balance = balance + :amount WHERE id = :id"
            await db.execute(update2, values={"id": to_id, "amount": amount})
```

## 5. **实际解决方案**

### **方案1：使用数据库事务**
```python
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, update

async def update_user(session: AsyncSession, user_id: int):
    async with session.begin():
        # 获取并锁定行
        stmt = select(User).where(User.id == user_id).with_for_update()
        result = await session.execute(stmt)
        user = result.scalar_one()
        
        # 修改数据
        user.balance += 100
        await session.commit()
```

### **方案2：使用分布式锁**
```python
import redis.asyncio as redis
import asyncio

class DistributedLock:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
    
    async def lock(self, key: str, timeout=10):
        import uuid
        identifier = str(uuid.uuid4())
        end = asyncio.get_event_loop().time() + timeout
        
        while asyncio.get_event_loop().time() < end:
            if await self.redis.setnx(key, identifier):
                await self.redis.expire(key, timeout)
                return identifier
            await asyncio.sleep(0.01)
        return None
    
    async def unlock(self, key: str, identifier: str):
        script = """
        if redis.call('get', KEYS[1]) == ARGV[1] then
            return redis.call('del', KEYS[1])
        else
            return 0
        end
        """
        await self.redis.eval(script, 1, key, identifier)

# 使用
redis_client = redis.Redis()
lock = DistributedLock(redis_client)

async def safe_update(user_id: int):
    lock_key = f"lock:user:{user_id}"
    lock_id = await lock.lock(lock_key)
    if lock_id:
        try:
            # 执行更新操作
            pass
        finally:
            await lock.unlock(lock_key, lock_id)
```

### **方案3：使用消息队列串行化**
```python
import asyncio
from collections import defaultdict

class TaskQueue:
    def __init__(self):
        self.queues = defaultdict(asyncio.Queue)
        self.locks = {}
    
    async def process_for_key(self, key: str, coro_func, *args):
        if key not in self.locks:
            self.locks[key] = asyncio.Lock()
        
        async with self.locks[key]:
            return await coro_func(*args)

queue = TaskQueue()

@app.post("/update/{user_id}")
async def update_user(user_id: int):
    async def _update():
        # 这个函数对同一个user_id会串行执行
        # 对不同user_id可以并行执行
        pass
    
    return await queue.process_for_key(f"user_{user_id}", _update)
```

## 6. **关键总结**

1. **asyncio 是单线程的**，真正的并发在 `await` 点切换
2. **竞态条件可能发生**：即使单线程，协程切换也可能导致数据不一致
3. **数据库层面需要额外保护**：
   - 使用事务和适当的隔离级别
   - 使用乐观锁或悲观锁
4. **应用层保护**：
   - 对共享资源使用锁（asyncio.Lock 用于单进程）
   - 使用分布式锁（跨进程/服务）
   - 使用消息队列串行化处理

## 7. **最佳实践建议**

```python
# 1. 尽量使用数据库的事务机制
# 2. 对关键操作使用行级锁或乐观锁
# 3. 避免在应用层维护共享状态
# 4. 如果必须，使用 asyncio.Lock（单进程）或分布式锁

import asyncio

# 应用层锁（仅限单进程）
lock = asyncio.Lock()

async def safe_operation():
    async with lock:  # 这会序列化所有访问
        # 操作共享资源
        pass
```

记住：**数据库的 ACID 特性是你的第一道防线**，应用层保护是补充。在设计系统时，优先考虑让数据库处理并发问题。

    

 
# **asyncio.Lock() 详细解析与多场景示例**

`asyncio.Lock()` 是 asyncio 中的互斥锁，用于在**单进程、单线程**的异步环境中保护共享资源。让我详细说明并提供多个例子。

## **1. 基本工作原理**

```python
import asyncio

lock = asyncio.Lock()  # 创建一个锁对象

async def critical_section():
    # 等待获取锁
    async with lock:  # 进入时自动获取锁
        # 临界区代码 - 同一时间只有一个协程能执行这里
        await asyncio.sleep(1)
        print("执行临界区代码")
    # 退出时自动释放锁

# 模拟并发
async def main():
    # 创建多个任务并发执行
    tasks = [critical_section() for _ in range(5)]
    await asyncio.gather(*tasks)

# 执行结果：虽然创建了5个任务，但它们会串行执行临界区代码
# 每个任务等待前一个释放锁后，才能进入临界区
```

## **2. 示例1：保护共享计数器**

### **问题场景（无锁时）：**
```python
import asyncio

counter = 0

async def unsafe_increment():
    global counter
    temp = counter  # 读取当前值
    await asyncio.sleep(0.01)  # 模拟IO操作，让出控制权
    counter = temp + 1  # 写回新值

async def main():
    global counter
    # 并发执行100次自增
    tasks = [unsafe_increment() for _ in range(100)]
    await asyncio.gather(*tasks)
    print(f"最终结果: {counter}")  # 可能不是100！
    
# 运行结果可能小于100，因为存在竞态条件
```

### **解决方案（使用锁）：**
```python
import asyncio

counter = 0
lock = asyncio.Lock()

async def safe_increment():
    global counter
    async with lock:  # 保护整个操作
        temp = counter
        await asyncio.sleep(0.01)  # 即使在await期间，锁也不会释放
        counter = temp + 1

async def main():
    global counter
    tasks = [safe_increment() for _ in range(100)]
    await asyncio.gather(*tasks)
    print(f"最终结果: {counter}")  # 保证是100

# asyncio.run(main())
```

## **3. 示例2：数据库连接池管理**

```python
import asyncio
from typing import List

class DatabaseConnectionPool:
    def __init__(self, pool_size: int = 5):
        self.pool_size = pool_size
        self.available = [f"conn_{i}" for i in range(pool_size)]
        self.in_use = []
        self.lock = asyncio.Lock()  # 保护连接池状态
    
    async def get_connection(self):
        async with self.lock:
            # 等待直到有可用连接
            while not self.available:
                print("无可用连接，等待中...")
                # 临时释放锁，让其他协程可以归还连接
                await self.lock.release()
                await asyncio.sleep(0.1)  # 短暂等待
                await self.lock.acquire()  # 重新获取锁
            
            conn = self.available.pop()
            self.in_use.append(conn)
            print(f"获取连接: {conn}, 可用: {self.available}, 使用中: {self.in_use}")
            return conn
    
    async def release_connection(self, conn: str):
        async with self.lock:
            if conn in self.in_use:
                self.in_use.remove(conn)
                self.available.append(conn)
                print(f"释放连接: {conn}, 可用: {self.available}, 使用中: {self.in_use}")

# 使用示例
async def query_database(pool: DatabaseConnectionPool, task_id: int):
    conn = await pool.get_connection()
    try:
        # 模拟数据库查询
        await asyncio.sleep(0.5)
        print(f"任务 {task_id} 使用连接 {conn} 执行查询")
    finally:
        await pool.release_connection(conn)

async def main():
    pool = DatabaseConnectionPool(pool_size=3)
    tasks = [query_database(pool, i) for i in range(10)]
    await asyncio.gather(*tasks)
```

## **4. 示例3：缓存系统**

```python
import asyncio
from datetime import datetime, timedelta

class AsyncCache:
    def __init__(self):
        self.cache = {}
        self.lock = asyncio.Lock()
    
    async def get(self, key: str, default=None):
        async with self.lock:
            if key in self.cache:
                entry = self.cache[key]
                if entry['expires_at'] > datetime.now():
                    return entry['value']
                else:
                    # 已过期，删除
                    del self.cache[key]
            return default
    
    async def set(self, key: str, value, ttl_seconds: int = 60):
        async with self.lock:
            self.cache[key] = {
                'value': value,
                'expires_at': datetime.now() + timedelta(seconds=ttl_seconds)
            }
    
    async def clear_expired(self):
        async with self.lock:
            now = datetime.now()
            expired_keys = [
                key for key, entry in self.cache.items()
                if entry['expires_at'] <= now
            ]
            for key in expired_keys:
                del self.cache[key]
            return len(expired_keys)

# 使用示例
cache = AsyncCache()

async def get_user_data(user_id: str):
    # 先尝试从缓存获取
    data = await cache.get(f"user:{user_id}")
    if data is not None:
        print(f"缓存命中: {user_id}")
        return data
    
    # 缓存未命中，模拟从数据库获取
    print(f"缓存未命中，查询数据库: {user_id}")
    await asyncio.sleep(0.5)  # 模拟数据库查询
    data = f"用户 {user_id} 的数据"
    
    # 写入缓存
    await cache.set(f"user:{user_id}", data, ttl_seconds=10)
    return data

async def main():
    # 模拟并发请求
    tasks = []
    for i in range(5):
        tasks.append(get_user_data("user1"))  # 相同key，应该只有一个真正查询数据库
        tasks.append(get_user_data(f"user{i}"))  # 不同key，会并发查询
    
    await asyncio.gather(*tasks)
```

## **5. 示例4：限流器（Rate Limiter）**

```python
import asyncio
import time

class RateLimiter:
    def __init__(self, requests_per_second: int):
        self.requests_per_second = requests_per_second
        self.timestamps = []
        self.lock = asyncio.Lock()
    
    async def acquire(self):
        async with self.lock:
            now = time.time()
            
            # 移除1秒前的记录
            cutoff = now - 1.0
            self.timestamps = [ts for ts in self.timestamps if ts > cutoff]
            
            # 检查是否超过限制
            if len(self.timestamps) >= self.requests_per_second:
                # 计算需要等待的时间
                oldest = self.timestamps[0]
                wait_time = oldest + 1.0 - now
                
                # 释放锁，等待指定时间
                await self.lock.release()
                try:
                    await asyncio.sleep(wait_time)
                finally:
                    await self.lock.acquire()  # 重新获取锁
                
                # 重新计算（因为等待期间可能有变化）
                now = time.time()
                self.timestamps = [ts for ts in self.timestamps if ts > now - 1.0]
            
            # 添加当前时间戳
            self.timestamps.append(now)

# 使用示例
limiter = RateLimiter(requests_per_second=3)  # 每秒最多3个请求

async def make_request(request_id: int):
    await limiter.acquire()  # 等待直到允许执行
    print(f"请求 {request_id} 在 {time.time():.2f} 执行")
    await asyncio.sleep(0.1)  # 模拟请求处理

async def main():
    # 模拟10个并发请求
    tasks = [make_request(i) for i in range(10)]
    await asyncio.gather(*tasks)
```

## **6. 示例5：工作队列**

```python
import asyncio
from collections import deque

class AsyncWorkQueue:
    def __init__(self, max_workers: int = 3):
        self.max_workers = max_workers
        self.queue = deque()
        self.active_workers = 0
        self.lock = asyncio.Lock()
        self.not_empty = asyncio.Condition(lock=self.lock)
        self.not_full = asyncio.Condition(lock=self.lock)
    
    async def put(self, item):
        async with self.lock:
            # 等待队列不满（如果有最大长度限制）
            self.queue.append(item)
            self.not_empty.notify()  # 通知有任务可处理
    
    async def get(self):
        async with self.lock:
            # 等待队列不空
            while not self.queue:
                await self.not_empty.wait()
            return self.queue.popleft()
    
    async def worker(self, worker_id: int):
        while True:
            try:
                # 获取任务
                task = await self.get()
                
                async with self.lock:
                    self.active_workers += 1
                
                # 执行任务
                print(f"Worker {worker_id} 开始处理任务: {task}")
                await asyncio.sleep(1)  # 模拟任务处理
                print(f"Worker {worker_id} 完成任务: {task}")
                
                async with self.lock:
                    self.active_workers -= 1
                    if self.active_workers < self.max_workers:
                        self.not_full.notify()  # 通知可以添加新任务
                        
            except asyncio.CancelledError:
                break
    
    async def start(self, num_workers: int = None):
        if num_workers is None:
            num_workers = self.max_workers
        
        workers = []
        for i in range(num_workers):
            worker_task = asyncio.create_task(self.worker(i))
            workers.append(worker_task)
        
        return workers

# 使用示例
async def main():
    queue = AsyncWorkQueue(max_workers=2)
    
    # 启动工作者
    workers = await queue.start()
    
    # 添加任务
    for i in range(10):
        await queue.put(f"任务{i}")
        print(f"已添加任务{i}")
        await asyncio.sleep(0.2)  # 控制添加速度
    
    # 等待所有任务完成
    await asyncio.sleep(10)
    
    # 清理
    for worker in workers:
        worker.cancel()
```

## **7. 重要注意事项**

### **锁的嵌套使用：**
```python
async def nested_locks_example():
    lock1 = asyncio.Lock()
    lock2 = asyncio.Lock()
    
    async with lock1:
        print("获取 lock1")
        await asyncio.sleep(0.1)
        
        # 注意：这可能导致死锁！
        async with lock2:
            print("获取 lock2")
            await asyncio.sleep(0.1)
```

### **避免死锁的建议：**
1. **固定获取顺序**：总是按相同顺序获取多个锁
2. **使用超时**：
```python
try:
    await asyncio.wait_for(lock.acquire(), timeout=5.0)
    # 执行操作
except asyncio.TimeoutError:
    print("获取锁超时")
finally:
    if lock.locked():
        lock.release()
```

3. **使用 `asyncio.Lock()` 而不是 `threading.Lock()`**

### **性能考虑：**
```python
# 不好的做法：锁的范围太大
async def bad_example():
    async with lock:
        # IO操作不需要在锁内
        data = await fetch_from_network()  # 这会阻塞其他协程
        process(data)

# 好的做法：只保护必要的部分
async def good_example():
    # IO操作在锁外
    data = await fetch_from_network()
    
    # 只保护共享资源的访问
    async with lock:
        shared_resource.update(data)
```

## **8. 总结**

`asyncio.Lock()` 的关键点：
1. **协程级别互斥**：保护单进程内的异步协程访问共享资源
2. **非阻塞等待**：等待锁时不会阻塞整个线程
3. **可重入**：同一个协程可以多次获取同一个锁（计数机制）
4. **需要配合 `async with` 使用**：确保锁的正确释放

**适用场景**：
- 保护内存中的共享数据结构
- 管理有限资源（连接池、线程池）
- 实现异步的同步原语（信号量、条件变量等）
- 单进程内的并发控制

**不适用场景**：
- 多进程环境（需要使用 `multiprocessing.Lock()`）
- 分布式系统（需要使用分布式锁如 Redis 锁）
- 数据库并发（应该使用数据库的事务和锁机制）

 
asyncio.gather 的作用是并发执行多个协程（任务），并等待它们全部完成。

```
async def main():
    # 模拟10个并发请求
    tasks = [make_request(i) for i in range(10)]
    await asyncio.gather(*tasks)
```

- **并发执行**   | 10个 `make_request` 同时启动，不是顺序执行 
- **等待全部完成** | 会阻塞直到最后一个任务结束                  
- **收集结果**   | 返回所有任务的返回值列表（按传入顺序）            
- **异常传播**   | 默认第一个异常会取消其他任务                 

对比：不用 vs 用 gather

 
```
# ❌ 顺序执行（慢）
for i in range(10):
    await make_request(i)  # 等1个完成再启动下一个，总耗时 = 10 × 单个耗时

# ✅ 并发执行（快）
tasks = [make_request(i) for i in range(10)]
await asyncio.gather(*tasks)  # 10个同时启动，总耗时 ≈ 最慢的那个
```

实际运行流程

 
```
import asyncio
import aiohttp

async def make_request(i):
    print(f"[{i}] 开始")
    await asyncio.sleep(1)  # 模拟网络IO
    print(f"[{i}] 完成")
    return f"结果{i}"

async def main():
    tasks = [make_request(i) for i in range(3)]
    results = await asyncio.gather(*tasks)
    print(f"所有结果: {results}")

asyncio.run(main())
```
[0] 开始
[1] 开始
[2] 开始
[0] 完成  (约1秒后)
[1] 完成
[2] 完成
所有结果: ['结果0', '结果1', '结果2']

常用参数

 
```
results = await asyncio.gather(
    *tasks,
    return_exceptions=True  # 异常不抛出，作为结果返回
)
```

在 FastAPI 中的典型应用

 
```
@app.get("/batch-users")
async def get_batch_users(user_ids: list[int]):
    # 并发查询多个用户，而不是循环 await
    tasks = [fetch_user(uid) for uid in user_ids]
    users = await asyncio.gather(*tasks)
    return users
```
一句话总结：gather 是 asyncio 的"并发执行 + 汇总结果"工具，让多个 IO 操作并行处理，大幅提升效率。