1. 项目概述最近在帮学弟调试一个课程设计项目时发现很多同学在使用Python操作MySQL数据库时存在不少误区。pymysql作为Python连接MySQL最常用的库之一虽然基础但藏着不少使用技巧。今天我就以读取表内容这个最基础的操作场景为例分享一些实战中总结的经验。2. 环境准备与基础配置2.1 安装与导入首先确保已安装pymysql库pip install pymysql在Python脚本中导入import pymysql注意如果同时使用pandas做数据处理建议安装pymysql的兼容版本避免出现数据类型转换问题。2.2 连接参数详解建立数据库连接时需要以下核心参数conn pymysql.connect( hostlocalhost, # 数据库服务器地址 userroot, # 用户名 password123456, # 密码 databasetest, # 数据库名 port3306, # 端口MySQL默认3306 charsetutf8mb4 # 字符编码 )参数选择建议charset强烈建议使用utf8mb4而非utf8支持完整的Unicode字符如emoji生产环境避免使用root账户应创建专用账户并限制权限连接超时时间可通过connect_timeout参数设置默认10秒3. 数据读取实战3.1 基础查询方法try: with conn.cursor() as cursor: # 执行SQL查询 sql SELECT * FROM users WHERE age %s cursor.execute(sql, (18,)) # 获取结果 results cursor.fetchall() for row in results: print(row) finally: conn.close()关键点说明使用with语句管理cursor确保资源释放SQL参数使用%s占位符避免SQL注入fetchall()返回全部结果大数据量时慎用3.2 高级查询技巧3.2.1 分页查询优化def query_with_pagination(page, page_size): offset (page - 1) * page_size sql SELECT * FROM large_table LIMIT %s OFFSET %s cursor.execute(sql, (page_size, offset)) return cursor.fetchall()性能提示大数据量分页避免使用OFFSET可改用WHERE id last_id方式3.2.2 流式读取处理海量数据时使用SSCursorwith conn.cursor(pymysql.cursors.SSCursor) as cursor: cursor.execute(SELECT * FROM huge_table) while True: row cursor.fetchone() if not row: break process_row(row)4. 异常处理与性能优化4.1 完善的错误处理try: conn pymysql.connect(...) with conn.cursor() as cursor: cursor.execute(...) except pymysql.MySQLError as e: print(f数据库错误: {e.args[0]} - {e.args[1]}) except Exception as e: print(f其他错误: {str(e)}) finally: if conn in locals() and conn.open: conn.close()常见错误代码2003: 连接拒绝1045: 访问被拒绝1146: 表不存在1213: 死锁4.2 连接池管理高并发场景建议使用连接池from pymysql import pools pool pools.Pool( hostlocalhost, userroot, password123456, databasetest, min2, # 最小连接数 max10 # 最大连接数 ) conn pool.get_conn() try: with conn.cursor() as cursor: cursor.execute(...) finally: pool.release(conn)5. 实战案例学生成绩分析系统假设我们需要从学生成绩表中统计各科平均分def analyze_scores(): conn pymysql.connect(...) try: with conn.cursor() as cursor: # 查询各科平均分 sql SELECT subject, AVG(score) as avg_score FROM student_scores GROUP BY subject ORDER BY avg_score DESC cursor.execute(sql) # 使用DictCursor获取字段名 results cursor.fetchall() for row in results: print(f{row[0]}: {row[1]:.2f}) finally: conn.close()优化建议添加适当索引如subject字段大数据表考虑分批处理复杂计算可下推到数据库执行6. 常见问题排查6.1 连接超时问题现象频繁出现OperationalError: (2013, Lost connection to MySQL server)解决方案增加connect_timeout参数值检查网络稳定性设置wait_timeout和interactive_timeout参数6.2 中文乱码问题确保三处编码一致连接参数charsetutf8mb4数据库/表字符集为utf8mb4Python文件头部添加# -- coding: utf-8 --6.3 事务隔离问题默认REPEATABLE-READ隔离级别可能导致幻读可根据需要调整conn pymysql.connect(..., autocommitFalse) conn.begin() # 开始事务 # 执行操作... conn.commit() # 或 conn.rollback()7. 扩展应用7.1 与Pandas集成import pandas as pd def query_to_dataframe(sql): conn pymysql.connect(...) try: return pd.read_sql(sql, conn) finally: conn.close()7.2 异步操作aiomysqlimport asyncio import aiomysql async def async_query(): conn await aiomysql.connect(...) async with conn.cursor() as cursor: await cursor.execute(SELECT * FROM async_table) return await cursor.fetchall()在实际项目中我发现合理设置fetch_size参数能显著提升大数据量查询性能。对于百万级数据表建议设置为1000-5000既能减少网络往返次数又不会占用过多内存。