FastAPI真香！从考研失败到用它重构Java后端服务的实战心得

去年三月，我坐在考研复试教室外面，手心全是汗。结果不出所料——挂了。那一刻真的挺懵的，感觉自己四年的努力都白费了。但生活总得继续，投了几个月简历后，终于在一家做区块链数据服务的创业公司找到了后端开发的工作。

说来好笑，入职第一天就被安排接手一个用Spring Boot写的老旧Java服务，性能堪忧不说，代码还写得像迷宫。产品经理天天催着加新功能，测试小姐姐看我的眼神都带着怜悯。就在上周五晚上加班到凌晨两点，服务器又崩了，运维大哥在群里@我说“兄弟，你这接口QPS才50就扛不住了？”

那一刻我真的想砸电脑。

但转念一想，反正都要重构，为什么不试试Python的FastAPI？毕竟我现在刷LeetCode、准备跳槽都是靠ChatGPT辅助，对新技术接受度贼高。而且我们团队前端用的是Vue3，后端如果能快速提供RESTful API，大家都能轻松不少。

为什么是FastAPI而不是继续用Java？

我知道很多人看到这里可能会说：“Java多稳啊，为什么要换？”说实话，我之前也是Java阵营的忠实粉丝。Spring Boot确实强大，生态完善，各种中间件集成得飞起。但问题是——太重了！

我们的业务场景其实很简单：接收前端传来的区块链交易数据，做一些验证和处理，然后存到数据库里。这种I/O密集型的任务，用Java简直就是杀鸡用牛刀。而且每次改个接口都要重新编译、打包、部署，开发效率低得让人发指。

FastAPI的优势就很明显了：

• 异步支持原生：Python 3.7+的async/await语法让异步编程变得简单
• 自动文档生成：Swagger UI和ReDoc开箱即用，前端同学再也不用追着我要API文档了
• 类型提示友好：Pydantic模型让数据验证变得超简单
• 性能接近Node.js：基于Starlette和Uvicorn，性能比传统的Flask/Django快很多

最重要的是，对于我们这种小团队来说，开发速度快才是王道。老板要的是功能，不是技术栈有多高大上。

从零开始搭建FastAPI项目

首先安装必要的依赖：

pip install fastapi uvicorn[standard] sqlalchemy pydantic python-dotenv

然后创建项目结构：

blockchain-api/
├── main.py              # 应用入口
├── models/              # 数据库模型
│   └── transaction.py
├── schemas/             # Pydantic模型
│   └── transaction.py
├── crud/                # 数据库操作
│   └── transaction.py
├── database.py          # 数据库连接
└── .env                 # 环境变量

核心应用配置

main.py的内容很简洁：

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from database import engine
from models import transaction

# 创建数据库表
transaction.Base.metadata.create_all(bind=engine)

app = FastAPI(
    title="Blockchain Transaction API",
    description="处理区块链交易数据的后端服务",
    version="1.0.0"
)

# 配置CORS（前端跨域问题终于不用求运维了！）
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 生产环境记得改成具体域名
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.get("/")
async def root():
    return {"message": "Welcome to Blockchain API!"}

看到没？就这么几行代码，一个基本的API服务就跑起来了。对比一下以前用Spring Boot，光是那些@Configuration、@ComponentScan注解就能写半页纸。

数据模型设计

因为我们处理的是区块链交易数据，所以数据库设计要考虑以下几点：

1. 高并发写入：区块链数据量巨大，需要优化写入性能
2. 数据完整性：交易哈希必须唯一
3. 查询效率：经常按时间范围、地址等条件查询

models/transaction.py：

from sqlalchemy import Column, Integer, String, DateTime, Numeric, Index
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime

Base = declarative_base()

class Transaction(Base):
    __tablename__ = "transactions"
    
    id = Column(Integer, primary_key=True, index=True)
    tx_hash = Column(String(66), unique=True, nullable=False)  # 0x开头的64位哈希
    from_address = Column(String(42), nullable=False)         # 以太坊地址格式
    to_address = Column(String(42), nullable=False)
    value = Column(Numeric(precision=38, scale=18), nullable=False)  # 支持大数
    timestamp = Column(DateTime, default=datetime.utcnow)
    block_number = Column(Integer, nullable=False)
    
    # 创建复合索引提高查询性能
    __table_args__ = (
        Index('idx_from_timestamp', 'from_address', 'timestamp'),
        Index('idx_to_timestamp', 'to_address', 'timestamp'),
        Index('idx_block_number', 'block_number'),
    )

Pydantic模型：类型安全的数据验证

这是FastAPI最让我惊艳的地方！以前用Java的时候，DTO、VO、BO各种对象转换，看得眼花缭乱。现在用Pydantic，几行代码就搞定：

schemas/transaction.py：

from pydantic import BaseModel, validator
from datetime import datetime
from decimal import Decimal

class TransactionCreate(BaseModel):
    tx_hash: str
    from_address: str
    to_address: str
    value: Decimal
    block_number: int
    
    @validator('tx_hash')
    def validate_tx_hash(cls, v):
        if not v.startswith('0x') or len(v) != 66:
            raise ValueError('Invalid transaction hash format')
        return v
    
    @validator('from_address', 'to_address')
    def validate_address(cls, v):
        if not v.startswith('0x') or len(v) != 42:
            raise ValueError('Invalid Ethereum address format')
        return v

class TransactionResponse(TransactionCreate):
    id: int
    timestamp: datetime
    
    class Config:
        orm_mode = True  # 允许从ORM模型直接序列化

前端传过来的数据会自动验证，不符合格式的直接返回422错误，再也不用担心脏数据进数据库了！

异步数据库操作：告别阻塞

说到异步，这可是FastAPI的杀手锏。我们的区块链数据服务经常需要同时处理成百上千的请求，如果用同步的方式，一个慢查询就会阻塞整个服务。

crud/transaction.py：

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from sqlalchemy import and_
from models.transaction import Transaction
from schemas.transaction import TransactionCreate
from datetime import datetime

async def create_transaction(db: AsyncSession, transaction: TransactionCreate):
    """创建交易记录"""
    db_transaction = Transaction(
        tx_hash=transaction.tx_hash,
        from_address=transaction.from_address,
        to_address=transaction.to_address,
        value=transaction.value,
        block_number=transaction.block_number,
        timestamp=datetime.utcnow()
    )
    db.add(db_transaction)
    await db.commit()
    await db.refresh(db_transaction)
    return db_transaction

async def get_transactions_by_address(
    db: AsyncSession, 
    address: str, 
    start_time: datetime = None,
    end_time: datetime = None,
    skip: int = 0, 
    limit: int = 100
):
    """根据地址查询交易记录"""
    query = select(Transaction).where(
        or_(
            Transaction.from_address == address,
            Transaction.to_address == address
        )
    )
    
    if start_time:
        query = query.where(Transaction.timestamp >= start_time)
    if end_time:
        query = query.where(Transaction.timestamp <= end_time)
    
    query = query.offset(skip).limit(limit).order_by(Transaction.timestamp.desc())
    result = await db.execute(query)
    return result.scalars().all()

注意这里用的是AsyncSession，配合Uvicorn的异步worker，性能提升非常明显。我在本地压测了一下，同样的机器配置，FastAPI版本的QPS能达到800+，而原来的Java服务只有50左右。

生产环境部署经验

别以为写完代码就万事大吉了，生产环境才是真正的考验。分享几个踩过的坑：

1. 数据库连接池配置

默认的连接池太小，在高并发下会报Too many connections错误。解决方案是在database.py中配置合适的连接池大小：

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
import os

DATABASE_URL = os.getenv("DATABASE_URL")

engine = create_async_engine(
    DATABASE_URL,
    echo=False,  # 生产环境关闭SQL日志
    pool_size=20,      # 连接池大小
    max_overflow=30,   # 超出pool_size后的最大连接数
    pool_pre_ping=True, # 检测连接是否有效
    pool_recycle=3600   # 1小时回收连接
)

async_session = sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)

2. 日志监控不能少

FastAPI本身不提供完善的日志系统，需要自己集成。我用了structlog，配合ELK栈，排查问题方便多了：

import structlog
import logging

# 配置结构化日志
structlog.configure(
    processors=[
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.stdlib.PositionalArgumentsFormatter(),
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer()
    ],
    context_class=dict,
    logger_factory=structlog.stdlib.LoggerFactory(),
    wrapper_class=structlog.stdlib.BoundLogger,
    cache_logger_on_first_use=True,
)

logger = structlog.get_logger()

3. 健康检查接口

运维大哥要求所有服务都要有健康检查接口，这个很简单：

@app.get("/health")
async def health_check():
    return {
        "status": "healthy",
        "timestamp": datetime.utcnow().isoformat(),
        "version": "1.0.0"
    }

性能对比：FastAPI vs Spring Boot

为了说服技术负责人同意重构，我做了详细的性能对比测试。测试环境：4核8G的云服务器，PostgreSQL 13，1000并发用户。

指标	FastAPI (异步)	Spring Boot (同步)	提升
平均响应时间	12ms	85ms	7x
QPS	820	48	17x
内存占用	120MB	650MB	5.4x
启动时间	0.8s	8.2s	10x

看到这个数据，技术负责人眼睛都亮了。特别是内存占用，对我们这种要部署多个微服务的小公司来说，简直是救命稻草。

给前端同学的福利：自动生成API文档

最让我感动的是前端小妹妹的反馈。以前她总是追着我要API文档，现在直接访问/docs就能看到完整的Swagger UI：

GET    /transactions/{address}    # 获取指定地址的交易记录
POST   /transactions             # 创建新交易记录  
GET    /health                   # 服务健康检查

每个接口都有详细的参数说明、请求示例、响应格式，还能直接在页面上测试。她说：“终于不用看你的Java代码猜接口参数了！”

写在最后

从考研失败到现在的后端开发，这一路走来真的不容易。但FastAPI让我重新找回了编程的乐趣——简洁、高效、直观。虽然很多人说Python不适合做后端，但在合适的场景下，它真的能发挥巨大的价值。

现在我已经用FastAPI重构了三个核心服务，性能提升了十几倍，代码量减少了60%。上周团建时，运维大哥还特意敬了我一杯：“兄弟，服务器费用省了不少啊！”

如果你也在考虑重构老旧的Java服务，或者想找一个快速开发API的框架，不妨试试FastAPI。当然，前提是你得先搞定产品经理的需求变更（手动狗头）。

对了，最近在刷题准备跳槽，如果有深圳的小伙伴想一起交流FastAPI或者云原生技术，欢迎私信！毕竟在这个卷成麻花的行业里，能有个技术搭子真的很重要。

---

本文写于2024年3月的一个深夜，刚fix完一个线上bug，咖啡已经凉了，但心里暖暖的——因为明天就要上线新版本了！