FastAPI状态管理：简单实现全局变量与缓存

from fastapi import FastAPI
import asyncio

app = FastAPI()

# 定义全局变量：用于存储计数器
count = 0
# 定义异步锁（解决多请求同时修改的竞态条件）
lock = asyncio.Lock()

@app.get("/counter")
async def get_counter():
    global count
    async with lock:  # 获取锁，确保同一时间只有一个请求修改count
        count += 1
        return {"current_count": count}

from fastapi import FastAPI
import time

app = FastAPI()

# 模拟数据库查询（实际项目中可替换为真实数据库操作）
def get_user_from_db(user_id: int):
    time.sleep(1)  # 模拟1秒查询延迟
    return {"user_id": user_id, "name": f"User_{user_id}"}

# 定义内存缓存字典
user_cache = {}

@app.get("/user/{user_id}")
async def get_user(user_id: int):
    # 1. 先查缓存
    if user_id in user_cache:
        return {"source": "cache", "data": user_cache[user_id]}

    # 2. 缓存未命中，从数据库查询并缓存
    user_data = get_user_from_db(user_id)
    user_cache[user_id] = user_data  # 存入缓存
    return {"source": "database", "data": user_data}

from fastapi import FastAPI
from cachetools import LRUCache, TTLCache
from typing import Optional

app = FastAPI()

# 方式1：LRU缓存（最近最少使用，最多存储10条数据）
user_cache_lru = LRUCache(maxsize=10)

# 方式2：TTL缓存（自动过期，10秒内未被访问的数据自动清除）
user_cache_ttl = TTLCache(maxsize=10, ttl=10)

# 模拟用户查询（与前例相同）
def get_user_from_db(user_id: int):
    time.sleep(1)
    return {"user_id": user_id, "name": f"User_{user_id}"}

@app.get("/user/lru/{user_id}")
async def get_user_lru(user_id: int):
    # 检查LRU缓存
    if user_id in user_cache_lru:
        return {"source": "lru_cache", "data": user_cache_lru[user_id]}
    user_data = get_user_from_db(user_id)
    user_cache_lru[user_id] = user_data
    return {"source": "database", "data": user_data}

@app.get("/user/ttl/{user_id}")
async def get_user_ttl(user_id: int):
    # 检查TTL缓存（10秒过期）
    if user_id in user_cache_ttl:
        return {"source": "ttl_cache", "data": user_cache_ttl[user_id]}
    user_data = get_user_from_db(user_id)
    user_cache_ttl[user_id] = user_data
    return {"source": "database", "data": user_data}

from fastapi import FastAPI
import redis.asyncio as redis
import json

app = FastAPI()

# 连接Redis（需提前安装redis：pip install redis）
redis_client = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)

@app.get("/user/redis/{user_id}")
async def get_user_redis(user_id: int):
    # 1. 尝试从Redis缓存读取
    cached_data = await redis_client.get(f"user:{user_id}")
    if cached_data:
        return {"source": "redis_cache", "data": json.loads(cached_data)}

    # 2. 缓存未命中，从数据库查询
    user_data = get_user_from_db(user_id)
    # 3. 存入Redis，设置10分钟过期
    await redis_client.setex(
        f"user:{user_id}",  # 键：user:1
        600,                # 过期时间：600秒（10分钟）
        json.dumps(user_data)  # 值：JSON序列化
    )
    return {"source": "database", "data": user_data}