PyMySQL 连接池实战：FastAPI 高并发场景下的数据库优化方案

在FastAPI等现代Web框架构建高并发服务时，数据库连接管理往往成为性能瓶颈的关键因素。传统的一次请求创建一个连接的方式，在100+并发场景下会导致连接风暴、响应延迟等问题。本文将深入探讨如何通过PyMySQL连接池技术实现高效、稳定的数据库访问。

1. 为什么需要数据库连接池？

当我们在开发FastAPI应用时，每个API请求都可能需要执行数据库操作。如果每次请求都新建一个数据库连接，会产生三个显著问题：

• 连接建立开销：每次TCP三次握手、MySQL权限验证等流程需要30-100ms
• 连接数爆炸：100并发意味着同时存在100个连接，可能超过MySQL的max_connections限制
• 资源浪费：大量空闲连接占用内存和服务端资源

连接池通过预先创建并复用连接，可以显著提升性能。实测数据显示：

2. PyMySQL连接池核心实现

我们将基于pymysqlpool实现一个生产级连接池管理类：

from pymysqlpool import ConnectionPool from contextlib import contextmanager import pymysql class MySQLPool: def __init__(self, config): self.pool_config = { 'host': config['host'], 'port': config.get('port', 3306), 'user': config['user'], 'password': config['password'], 'database': config['database'], 'charset': config.get('charset', 'utf8mb4'), 'autocommit': config.get('autocommit', True), 'max_size': config.get('max_connections', 10), 'min_size': config.get('min_connections', 2), 'ping_check': config.get('ping_check', True) } self.pool = ConnectionPool(**self.pool_config) @contextmanager def get_connection(self): conn = self.pool.get_connection() try: yield conn finally: conn.close() def execute_query(self, sql, args=None): with self.get_connection() as conn: with conn.cursor(pymysql.cursors.DictCursor) as cursor: cursor.execute(sql, args or ()) return cursor.fetchall() def execute_update(self, sql, args=None): with self.get_connection() as conn: with conn.cursor() as cursor: affected_rows = cursor.execute(sql, args or ()) if not self.pool_config['autocommit']: conn.commit() return affected_rows

关键设计要点：

1. 连接预热：初始化时创建min_size个连接，避免首次请求等待
2. 健康检查：ping_check确保连接可用性
3. 上下文管理：自动归还连接，避免泄漏
4. 类型支持：DictCursor返回字典形式结果更易用

3. 在FastAPI中的集成实践

下面展示如何在FastAPI应用中全局管理连接池：

from fastapi import FastAPI, Depends from pymysql.cursors import DictCursor app = FastAPI() mysql_pool = MySQLPool({ 'host': '127.0.0.1', 'user': 'api_user', 'password': 'secure_password', 'database': 'ecommerce' }) @app.on_event("startup") async def startup_event(): # 初始化连接池 mysql_pool.pool.init() @app.on_event("shutdown") async def shutdown_event(): # 优雅关闭连接池 mysql_pool.pool.close() async def get_db(): with mysql_pool.get_connection() as conn: yield conn @app.get("/products") async def list_products(page: int = 1, size: int = 10): offset = (page - 1) * size sql = "SELECT * FROM products WHERE status=1 LIMIT %s OFFSET %s" return mysql_pool.execute_query(sql, (size, offset)) @app.post("/orders") async def create_order(order_data: dict): sql = """INSERT INTO orders (user_id, product_id, quantity) VALUES (%(user_id)s, %(product_id)s, %(quantity)s)""" return {"order_id": mysql_pool.execute_update(sql, order_data)}

4. 性能调优与压测

我们使用Locust进行压力测试，模拟100并发用户持续访问：

from locust import HttpUser, task, between class ApiUser(HttpUser): wait_time = between(0.1, 0.5) @task def load_products(self): self.client.get("/products") @task(3) def create_order(self): self.client.post("/orders", json={ "user_id": 1, "product_id": 42, "quantity": 2 })

关键调优参数建议：

# MySQL服务端配置 max_connections = 200 wait_timeout = 600 max_user_connections = 100 # 连接池配置 max_size = 50 # 最大连接数 min_size = 5 # 最小保持连接数 idle_timeout = 300 # 空闲连接超时(秒)

压测结果对比：

5. 生产环境注意事项

在实际部署时，还需要考虑以下关键点：

连接泄漏检测

# 定期检查连接未归还情况 def check_leaks(): active = mysql_pool.pool._used_connections if active > 0: logger.warning(f"Potential connection leaks: {active}")

动态扩容策略

# 根据负载动态调整连接池大小 def adjust_pool_size(): avg_wait = mysql_pool.pool.average_wait_time if avg_wait > 0.1: # 等待时间超过100ms new_size = min( mysql_pool.pool.max_size * 1.5, MAX_ALLOWED_CONNECTIONS ) mysql_pool.pool.resize(new_size)

监控指标暴露

# HELP mysql_connection_pool_size Current connection pool size # TYPE mysql_connection_pool_size gauge mysql_connection_pool_size{env="prod"} 15 # HELP mysql_connection_wait_time_seconds Connection wait time # TYPE mysql_connection_wait_time_seconds histogram mysql_connection_wait_time_seconds_bucket{le="0.05"} 1234

在Kubernetes环境中，建议通过HPA实现自动扩缩容：

apiVersion: autoscaling/v2 kind: HorizontalPodAutoscaler metadata: name: api-autoscaler spec: scaleTargetRef: apiVersion: apps/v1 kind: Deployment name: api-service minReplicas: 3 maxReplicas: 10 metrics: - type: Resource resource: name: cpu target: type: Utilization averageUtilization: 60 - type: External external: metric: name: mysql_connection_wait_time_seconds selector: matchLabels: env: prod target: type: AverageValue averageValue: 100ms

通过本文介绍的技术方案，我们在实际项目中成功将数据库瓶颈服务的吞吐量从800RPS提升到了4500RPS，同时保持了99.9%的可用性。连接池作为基础架构组件，其稳定性和性能直接影响整个系统的表现，值得投入精力进行精细调优。