Python生产环境常见 Bug 排查与修复
·
1 环境与依赖问题(占 80% 项目故障)
Bug 1:Python 版本不一致导致"在我电脑能跑"
现象:本地 Python 3.11 正常运行,服务器 Python 3.8 报错。
根因:f-string 增强语法、typing 新特性、match-case 等在低版本不兼容。
修复方案:
# 使用 pyenv 管理多版本
pyenv install 3.11.9
pyenv local 3.11.9
# 项目 README 明确声明
# Python >= 3.10
# 运行时版本检查(防御性编程)
import sys
if sys.version_info < (3, 10):
raise RuntimeError("本项目需要 Python 3.10 或更高版本")
Bug 2:依赖地狱——升级一个库导致另一个库崩溃
根因:直接 pip install 不锁定版本,传递依赖冲突。
修复方案:
# 使用 pip-tools 锁定依赖
pip install pip-tools
pip-compile requirements.in # 生成精确版本的 requirements.txt
pip-sync requirements.txt # 同步环境到锁定版本
# requirements.in(只写直接依赖) fastapi uvicorn[standard] sqlalchemy>=2.0 redis[hiredis]
Bug 3:pip 安装超时或失败
根因:默认 PyPI 源在国内网络不稳定。
修复方案:
pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple
pip config set global.trusted-host pypi.tuna.tsinghua.edu.cn
2 异常处理与日志问题
Bug 4:try/except 滥用导致错误被静默吞掉
错误写法:
# 危险:吞掉所有异常,线上问题无法定位
try:
process_payment(order)
except:
pass
正确写法:
import logging
logger = logging.getLogger(__name__)
try:
process_payment(order)
except PaymentGatewayError as e:
logger.error("支付网关异常", extra={
"order_id": order.id,
"error": str(e),
"gateway": order.gateway
})
raise # 重新抛出,让上层决定如何处理
except ValueError as e:
logger.warning("订单数据校验失败: %s", e)
return {"error": "invalid_order_data"}
Bug 5:生产环境用 print 调试,无结构化日志
根因:print 无法区分级别、无时间戳、无法写入文件。
修复方案:
import logging
import json
from datetime import datetime
# 结构化日志配置
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s | %(levelname)-8s | %(name)s | %(message)s',
handlers=[
logging.FileHandler("app.log"),
logging.StreamHandler()
]
)
# 推荐使用 structlog 实现 JSON 格式日志
import structlog
structlog.configure(
processors=[
structlog.stdlib.add_log_level,
structlog.processors.TimeStamper(fmt="iso"),
structlog.processors.JSONRenderer()
]
)
logger = structlog.get_logger()
logger.info("user_login", user_id=12345, ip="192.168.1.1", method="oauth2")
3 性能问题
Bug 6:for 循环中频繁访问数据库
错误写法:
# N+1 查询问题
users = db.query(User).all()
for user in users:
orders = db.query(Order).filter(Order.user_id == user.id).all() # 每次循环一次查询
print(f"{user.name}: {len(orders)} 个订单")
修复方案:
# 使用 JOIN 或预加载
from sqlalchemy.orm import joinedload
users = db.query(User).options(joinedload(User.orders)).all()
for user in users:
print(f"{user.name}: {len(user.orders)} 个订单")
Bug 7:I/O 阻塞导致接口响应慢
根因:同步调用外部 API、数据库查询阻塞事件循环。
修复方案:
import asyncio
import aiohttp
async def fetch_all(urls: List[str]) -> List[dict]:
"""并发请求多个外部 API"""
async with aiohttp.ClientSession() as session:
tasks = [fetch_one(session, url) for url in urls]
return await asyncio.gather(*tasks, return_exceptions=True)
async def fetch_one(session: aiohttp.ClientSession, url: str) -> dict:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
return await resp.json()
4 并发与多进程问题
Bug 8:多线程没提速(GIL 问题)
根因:Python GIL 导致 CPU 密集型任务多线程无效。
修复方案:
from concurrent.futures import ProcessPoolExecutor
import multiprocessing as mp
# CPU 密集型 → 多进程
def cpu_intensive_task(data):
return sum(i * i for i in range(data))
with ProcessPoolExecutor(max_workers=mp.cpu_count()) as executor:
results = list(executor.map(cpu_intensive_task, large_dataset))
# I/O 密集型 → 异步或线程池
from concurrent.futures import ThreadPoolExecutor
with ThreadPoolExecutor(max_workers=20) as executor:
results = list(executor.map(fetch_url, url_list))
Bug 9:多进程下全局变量失效
根因:每个子进程有独立内存空间,全局变量不共享。
修复方案:
# 使用 multiprocessing.Manager 共享状态
from multiprocessing import Manager, Process
def worker(shared_dict, key, value):
shared_dict[key] = value
if __name__ == "__main__":
with Manager() as manager:
shared = manager.dict()
processes = [
Process(target=worker, args=(shared, f"key_{i}", i))
for i in range(10)
]
for p in processes:
p.start()
for p in processes:
p.join()
print(dict(shared)) # 所有进程的写入都可见
5 数据库相关坑
Bug 10:数据库连接泄漏
现象:运行一段时间后报 Too many connections。
根因:未正确关闭数据库连接或 Session。
修复方案:
from contextlib import contextmanager
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
engine = create_engine(
"postgresql://user:pass@localhost/db",
pool_size=10,
max_overflow=20,
pool_pre_ping=True, # 连接前检测有效性
pool_recycle=3600 # 1小时回收连接
)
SessionLocal = sessionmaker(bind=engine)
@contextmanager
def get_db():
"""确保 Session 正确关闭"""
db = SessionLocal()
try:
yield db
db.commit()
except Exception:
db.rollback()
raise
finally:
db.close()更多推荐
所有评论(0)