构建高并发服务一直是程序员的痛点。咱们今天就用FastAPI和Redis组合来搞定这个难题！这套方案不仅能处理高并发请求，还能让系统性能暴增20倍。写这篇文章的时候我正在处理一个每秒1万并发的项目，这些经验都是实打实的。
FastAPI简介
FastAPI就像是Django的那个调皮但是成绩特别好的弟弟。它基于Python 3.6+的类型提示构建，运行速度贼快，写起来还特别舒服。
来看个最基础的例子：
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def read_root():
return {"Hello": "World"}
看到没？就这几行代码就能跑起来一个API服务，还自带超nice的交互文档。
温馨提示：别忘了用uvicorn启动服务哦，直接用python运行是不行的！
Redis来帮忙
单靠FastAPI还不够猛，得加上Redis这个神器。Redis就像是一个超级快的临时仓库，把常用的数据放在这里，查询速度能快得飞起。
整一个简单的缓存示例：
import aioredis
from fastapi import FastAPI
app = FastAPI()
# 创建Redis连接池
redis = aioredis.from_url('redis://localhost')
@app.get("/user/{user_id}")
async def get_user(user_id: int):
# 先看Redis里有没有
cache_key = f"user:{user_id}"
cached_user = await redis.get(cache_key)
if cached_user:
return {"user": cached_user, "source": "cache"}
# 假装这是从数据库查的
user_data = {"id": user_id, "name": "张三"}
# 塞进Redis
await redis.set(cache_key, str(user_data), ex=3600)
return {"user": user_data, "source": "db"}
高并发黑魔法
光有缓存还不够，想要扛住高并发，还得整点花活。
from fastapi import FastAPI, Depends
from fastapi.middleware.cors import CORSMiddleware
from redis.asyncio import ConnectionPool, Redis
import asyncio
app = FastAPI()
# 配置CORS，这很重要！
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
# Redis连接池配置
REDIS_POOL = ConnectionPool(
host='localhost',
port=6379,
db=0,
max_connections=10_000 # 关键参数！
)
async def get_redis() -> Redis:
redis = Redis(connection_pool=REDIS_POOL)
try:
yield redis
finally:
await redis.close()
@app.get("/concurrent_test")
async def test_concurrent(redis: Redis = Depends(get_redis)):
# 用Redis计数器实现限流
key = "request_count"
await redis.incr(key)
# 模拟业务处理
await asyncio.sleep(0.1)
count = await redis.get(key)
return {"current_requests": int(count)}
温馨提示：Redis连接池的max_connections参数特别重要，设太小了扛不住并发，设太大了服务器受不了，得根据实际情况调整。
性能优化小技巧
- 用Redis Pipeline批量处理：
@app.post("/batch_process")
async def process_batch(items: list, redis: Redis = Depends(get_redis)):
pipe = redis.pipeline()
for item in items:
pipe.set(f"item:{item['id']}", str(item))
await pipe.execute()
return {"status": "success"}
- 合理设置缓存失效时间：
# 热点数据5分钟过期
await redis.set("hot_data", data, ex=300)
# 非热点数据1小时过期
await redis.set("cold_data", data, ex=3600)
面对高并发，还有个大杀器 - Redis集群。不过这玩意配置起来有点复杂，感兴趣的可以去翻翻官方文档。
写完感觉还有好多可以说的，这个FastAPI+Redis的组合是真的猛。性能提升20倍是保守估计，如果你的业务场景主要是读操作，提升个几十倍都是很正常的事情。
贴心提醒：代码部署到生产环境前，记得修改Redis的密码，别像我一开始那样，差点被黑客大佬打穿！