Python MySQL事务实战：从转账异常到数据一致性，手把手教你避坑

Python MySQL事务实战：从转账异常到数据一致性，手把手教你避坑

我刚开始用Python操作MySQL时，最怕的就是转账扣款这种业务——钱扣了但没到账，或者重复扣款。直到真正理解了事务，才发现原来数据一致性可以这么简单！

引言：为什么你的转账代码总出问题？

你有没有遇到过这样的场景：用户下单支付，钱从账户扣了，但订单状态没更新？或者更糟，系统崩溃后数据处于“半完成”状态？这些问题的根源，就是事务管理没做好。

在Python MySQL开发中，事务是保证数据一致性的核心机制。今天，我将用10年实战经验，带你从零掌握MySQL事务管理，让你写的转账、库存扣减等关键业务代码，再也不会出现数据错乱。

学完本文你将掌握：

1. 事务的ACID原理（用大白话讲清楚）
2. 用pymysql实现完整的事务控制
3. 异常处理的正确姿势（避免事务泄露）
4. 事务隔离级别的实战选择
5. 一个完整的银行转账案例

一、环境准备：搭建你的第一个事务测试环境

1.1 安装必要的库

# 安装pymysql - 我们选择纯Python实现的驱动pipinstallpymysql# 安装mysql客户端（可选，用于验证）# brew install mysql-client # Mac# apt-get install mysql-client # Ubuntu

1.2 创建测试数据库和表

我们先创建一个简单的银行账户系统来模拟真实业务：

-- 创建数据库CREATEDATABASEIFNOTEXISTSbank_systemDEFAULTCHARSETutf8mb4;-- 使用数据库USEbank_system;-- 创建账户表CREATETABLEaccounts(idINTPRIMARYKEYAUTO_INCREMENT,account_noVARCHAR(20)UNIQUENOTNULLCOMMENT'账号',account_nameVARCHAR(50)NOTNULLCOMMENT'账户名',balanceDECIMAL(10,2)DEFAULT0.00COMMENT'余额',created_atTIMESTAMPDEFAULTCURRENT_TIMESTAMP)ENGINE=InnoDBDEFAULTCHARSET=utf8mb4;-- 创建交易记录表CREATETABLEtransactions(idINTPRIMARYKEYAUTO_INCREMENT,from_accountVARCHAR(20)COMMENT'转出账号',to_accountVARCHAR(20)COMMENT'转入账号',amountDECIMAL(10,2)NOTNULLCOMMENT'金额',transaction_typeENUM('TRANSFER','DEPOSIT','WITHDRAW')NOTNULL,statusENUM('SUCCESS','FAILED','PENDING')DEFAULT'PENDING',created_atTIMESTAMPDEFAULTCURRENT_TIMESTAMP,INDEXidx_account(from_account,to_account))ENGINE=InnoDBDEFAULTCHARSET=utf8mb4;-- 插入测试数据INSERTINTOaccounts(account_no,account_name,balance)VALUES('1001','张三',10000.00),('1002','李四',5000.00),('1003','王五',2000.00);

技巧提示：一定要用InnoDB引擎！只有InnoDB支持完整的事务特性，MyISAM不支持事务。

二、基础概念：事务到底是什么？

2.1 事务的ACID原则（用大白话解释）

原子性（Atomicity）：就像开关灯，要么开（全部成功），要么关（全部失败），没有中间状态。转账时，扣款和加款必须同时成功或同时失败。

一致性（Consistency）：转账前后，系统总金额必须保持不变。张三10000元，李四5000元，转账后总额还是15000元。

隔离性（Isolation）：多个用户同时转账时，互相看不到对方的中间状态。你转账时，别人查余额看到的是转账前或转账后的状态，不会看到“钱扣了但没到账”的中间态。

持久性（Durability）：一旦转账成功，即使服务器断电，数据也不会丢失。

2.2 为什么Python开发者需要事务？

我刚开始写Python数据库代码时，经常这样写：

# ❌ 错误示范：没有事务的转账代码deftransfer_without_transaction(from_account,to_account,amount):# 扣款cursor.execute("UPDATE accounts SET balance = balance - %s WHERE account_no = %s",(amount,from_account))# 这里如果程序崩溃...time.sleep(1)# 模拟网络延迟或程序异常# 加款cursor.execute("UPDATE accounts SET balance = balance + %s WHERE account_no = %s",(amount,to_account))

问题很明显：如果程序在两次UPDATE之间崩溃，钱扣了但没到账，数据就不一致了！

三、实战演练：用pymysql实现完整事务控制

3.1 基础事务操作（自动提交 vs 手动提交）

MySQL默认是自动提交模式，每个SQL语句都是一个独立事务。我们需要改为手动控制：

importpymysqlimportlogging# 配置日志，方便调试logging.basicConfig(level=logging.INFO)logger=logging.getLogger(__name__)classBankSystem:def__init__(self):self.connection=Nonedefconnect(self):"""建立数据库连接"""try:self.connection=pymysql.connect(host='localhost',user='root',password='your_password',database='bank_system',charset='utf8mb4',cursorclass=pymysql.cursors.DictCursor# 返回字典格式)logger.info("数据库连接成功")returnTrueexceptpymysql.Errorase:logger.error(f"数据库连接失败:{e}")returnFalsedefbasic_transfer(self,from_account,to_account,amount):"""基础转账函数 - 演示事务的基本用法"""cursor=Nonetry:# 1. 获取游标cursor=self.connection.cursor()# 2. 关闭自动提交，开始事务self.connection.autocommit(False)# 或者用 self.connection.begin()logger.info("开始事务，自动提交已关闭")# 3. 检查转出账户余额cursor.execute("SELECT balance FROM accounts WHERE account_no = %s FOR UPDATE",(from_account,))result=cursor.fetchone()ifnotresult:raiseValueError(f"账户{from_account}不存在")current_balance=result['balance']ifcurrent_balance<amount:raiseValueError(f"账户{from_account}余额不足，当前余额:{current_balance}")# 4. 执行转账操作# 扣款cursor.execute("UPDATE accounts SET balance = balance - %s WHERE account_no = %s",(amount,from_account))logger.info(f"从账户{from_account}扣款{amount}元")# 加款cursor.execute("UPDATE accounts SET balance = balance + %s WHERE account_no = %s",(amount,to_account))logger.info(f"向账户{to_account}加款{amount}元")# 记录交易cursor.execute(""" INSERT INTO transactions (from_account, to_account, amount, transaction_type, status) VALUES (%s, %s, %s, 'TRANSFER', 'SUCCESS') """,(from_account,to_account,amount))# 5. 提交事务self.connection.commit()logger.info("事务提交成功")returnTrueexceptExceptionase:# 6. 发生异常，回滚事务ifself.connection:self.connection.rollback()logger.warning(f"事务回滚:{e}")# 记录失败交易ifcursor:cursor.execute(""" INSERT INTO transactions (from_account, to_account, amount, transaction_type, status) VALUES (%s, %s, %s, 'TRANSFER', 'FAILED') """,(from_account,to_account,amount))self.connection.commit()returnFalsefinally:# 7. 恢复自动提交并关闭游标ifself.connection:self.connection.autocommit(True)ifcursor:cursor.close()# 测试代码if__name__=="__main__":bank=BankSystem()ifbank.connect():# 测试正常转账success=bank.basic_transfer('1001','1002',1000)print(f"转账结果:{'成功'ifsuccesselse'失败'}")# 测试余额不足success=bank.basic_transfer('1003','1001',5000)print(f"转账结果:{'成功'ifsuccesselse'失败'}")

关键点解析：

1. autocommit(False)：关闭自动提交，开始事务
2. FOR UPDATE：加行锁，防止其他事务修改
3. commit()：所有操作成功，提交事务
4. rollback()：任何一步失败，回滚所有操作
5. finally块：确保资源释放，避免连接泄露

3.2 事务的保存点（复杂事务的后悔药）

对于复杂的多步骤事务，我们可以设置保存点，实现部分回滚：

defcomplex_transfer_with_savepoint(self,from_account,to_account,amount,fee=0):"""带保存点的复杂转账（含手续费）"""cursor=Nonesavepoint_name="before_fee_deduction"try:cursor=self.connection.cursor()self.connection.autocommit(False)# 步骤1：检查余额并扣款cursor.execute("SELECT balance FROM accounts WHERE account_no = %s FOR UPDATE",(from_account,))result=cursor.fetchone()ifnotresult:raiseValueError(f"账户{from_account}不存在")total_deduction=amount+feeifresult['balance']<total_deduction:raiseValueError(f"余额不足，需要{total_deduction}，当前{result['balance']}")# 扣款cursor.execute("UPDATE accounts SET balance = balance - %s WHERE account_no = %s",(amount,from_account))# 设置保存点cursor.execute(f"SAVEPOINT{savepoint_name}")logger.info(f"保存点 '{savepoint_name}' 已创建")try:# 步骤2：扣除手续费（可能失败）iffee>0:cursor.execute("UPDATE accounts SET balance = balance - %s WHERE account_no = %s",(fee,from_account))logger.info(f"手续费{fee}元已扣除")# 步骤3：加款cursor.execute("UPDATE accounts SET balance = balance + %s WHERE account_no = %s",(amount,to_account))# 记录交易cursor.execute(""" INSERT INTO transactions (from_account, to_account, amount, transaction_type, status) VALUES (%s, %s, %s, 'TRANSFER', 'SUCCESS') """,(from_account,to_account,amount))self.connection.commit()returnTrueexceptExceptionasfee_error:# 只回滚到保存点，保留主扣款操作cursor.execute(f"ROLLBACK TO SAVEPOINT{savepoint_name}")logger.warning(f"手续费处理失败，回滚到保存点:{fee_error}")# 继续完成转账（不含手续费）cursor.execute("UPDATE accounts SET balance = balance + %s WHERE account_no = %s",(amount,to_account))cursor.execute(""" INSERT INTO transactions (from_account, to_account, amount, transaction_type, status) VALUES (%s, %s, %s, 'TRANSFER', 'SUCCESS') """,(from_account,to_account,amount))self.connection.commit()returnTrue# 转账成功，只是没收手续费exceptExceptionase:self.connection.rollback()logger.error(f"转账失败:{e}")returnFalsefinally:ifself.connection:self.connection.autocommit(True)ifcursor:cursor.close()

四、事务隔离级别：解决并发问题

4.1 四种隔离级别对比

4.2 实战：设置和验证隔离级别

defdemonstrate_isolation_levels(self):"""演示不同隔离级别的影响"""# 设置隔离级别为 READ COMMITTEDwithself.connection.cursor()ascursor:cursor.execute("SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED")cursor.execute("SELECT @@transaction_isolation")result=cursor.fetchone()print(f"当前隔离级别:{result['@@transaction_isolation']}")# 测试不可重复读问题print("\n=== 测试不可重复读 ===")deftransaction1():"""事务1：两次读取同一数据"""conn=pymysql.connect(host='localhost',user='root',password='your_password',database='bank_system',charset='utf8mb4')conn.autocommit(False)withconn.cursor()ascur:# 第一次读取cur.execute("SELECT balance FROM accounts WHERE account_no = '1001'")balance1=cur.fetchone()['balance']print(f"事务1第一次读取:{balance1}")# 等待事务2修改数据importtime time.sleep(2)# 第二次读取（在READ COMMITTED下，这里会读到新值）cur.execute("SELECT balance FROM accounts WHERE account_no = '1001'")balance2=cur.fetchone()['balance']print(f"事务1第二次读取:{balance2}")ifbalance1!=balance2:print("⚠️ 发生不可重复读！两次读取结果不同")conn.commit()conn.close()deftransaction2():"""事务2：修改数据"""importtime time.sleep(1)# 确保事务1先开始conn=pymysql.connect(host='localhost',user='root',password='your_password',database='bank_system',charset='utf8mb4')withconn.cursor()ascur:cur.execute("UPDATE accounts SET balance = balance + 100 WHERE account_no = '1001'")conn.commit()print("事务2已修改数据")conn.close()# 启动两个线程模拟并发importthreading t1=threading.Thread(target=transaction1)t2=threading.Thread(target=transaction2)t1.start()t2.start()t1.join()t2.join()

常见问题：

• 脏读：读到其他事务未提交的数据
• 不可重复读：同一事务内两次读取结果不同
• 幻读：同一查询条件，第二次看到新插入的行

五、异常处理的最佳实践

5.1 完整的异常处理模板

fromcontextlibimportcontextmanagerfromtypingimportOptional,Tuple@contextmanagerdeftransaction_context(connection,isolation_level:Optional[str]=None):""" 事务上下文管理器 使用示例： with transaction_context(conn) as cursor: cursor.execute("UPDATE ...") """cursor=Noneoriginal_autocommit=Noneoriginal_isolation=Nonetry:# 保存原始设置original_autocommit=connection.get_autocommit()ifisolation_level:withconnection.cursor()astemp_cursor:temp_cursor.execute("SELECT @@transaction_isolation")original_isolation=temp_cursor.fetchone()['@@transaction_isolation']temp_cursor.execute(f"SET SESSION TRANSACTION ISOLATION LEVEL{isolation_level}")# 开始事务connection.autocommit(False)cursor=connection.cursor()yieldcursor# 提交事务connection.commit()exceptExceptionase:# 回滚事务ifconnection:connection.rollback()logger.error(f"事务执行失败:{e}")raise# 重新抛出异常finally:# 恢复原始设置ifconnection:iforiginal_autocommitisnotNone:connection.autocommit(original_autocommit)ifisolation_levelandoriginal_isolation:withconnection.cursor()astemp_cursor:temp_cursor.execute(f"SET SESSION TRANSACTION ISOLATION LEVEL{original_isolation}")ifcursor:cursor.close()# 使用示例defsafe_transfer(self,from_account:str,to_account:str,amount:float)->Tuple[bool,str]:"""安全的转账函数，使用上下文管理器"""try:withtransaction_context(self.connection,"REPEATABLE READ")ascursor:# 检查余额cursor.execute("SELECT balance FROM accounts WHERE account_no = %s FOR UPDATE",(from_account,))result=cursor.fetchone()ifnotresult:returnFalse,f"账户{from_account}不存在"ifresult['balance']<amount:returnFalse,f"余额不足，当前余额:{result['balance']}"# 执行转账cursor.execute("UPDATE accounts SET balance = balance - %s WHERE account_no = %s",(amount,from_account))cursor.execute("UPDATE accounts SET balance = balance + %s WHERE account_no = %s",(amount,to_account))# 记录日志cursor.execute(""" INSERT INTO transactions (from_account, to_account, amount, transaction_type, status) VALUES (%s, %s, %s, 'TRANSFER', 'SUCCESS') """,(from_account,to_account,amount))returnTrue,"转账成功"exceptpymysql.Errorasdb_error:logger.error(f"数据库错误:{db_error}")returnFalse,f"数据库错误:{db_error}"exceptExceptionase:logger.error(f"系统错误:{e}")returnFalse,f"系统错误:{e}"

5.2 常见异常及处理方案

defdeadlock_retry_transfer(self,from_account,to_account,amount,max_retries=3):"""带死锁重试的转账函数"""forattemptinrange(max_retries):try:returnself.safe_transfer(from_account,to_account,amount)exceptpymysql.err.OperationalErrorase:# 错误码1213表示死锁ife.args[0]==1213:wait_time=(attempt+1)*0.5# 指数退避logger.warning(f"检测到死锁，第{attempt+1}次重试，等待{wait_time}秒")importtime time.sleep(wait_time)continueelse:raise# 其他错误直接抛出exceptExceptionase:raisereturnFalse,"转账失败，超过最大重试次数"

六、实际项目应用：电商订单系统

让我们看一个真实的电商订单创建场景：

classOrderSystem:defcreate_order(self,user_id:int,items:list,shipping_address:str)->dict:""" 创建订单 - 完整的事务示例 涉及：库存扣减、订单创建、支付记录 """result={"success":False,"order_id":None,"message":"","errors":[]}try:withtransaction_context(self.connection)ascursor:# 1. 验证并锁定库存total_amount=0foriteminitems:product_id=item['product_id']quantity=item['quantity']# 使用SELECT ... FOR UPDATE锁定库存行cursor.execute(""" SELECT stock, price FROM products WHERE id = %s AND status = 'ACTIVE' FOR UPDATE """,(product_id,))product=cursor.fetchone()ifnotproduct:raiseValueError(f"商品{product_id}不存在或已下架")ifproduct['stock']<quantity:raiseValueError(f"商品{product_id}库存不足，剩余:{product['stock']}")total_amount+=product['price']*quantity# 2. 扣减库存foriteminitems:cursor.execute(""" UPDATE products SET stock = stock - %s WHERE id = %s AND stock >= %s """,(item['quantity'],item['product_id'],item['quantity']))affected_rows=cursor.rowcountifaffected_rows==0:raiseValueError(f"商品{item['product_id']}库存扣减失败")# 3. 创建订单cursor.execute(""" INSERT INTO orders (user_id, total_amount, status, shipping_address) VALUES (%s, %s, 'PENDING', %s) """,(user_id,total_amount,shipping_address))order_id=cursor.lastrowid# 4. 创建订单明细foriteminitems:cursor.execute(""" SELECT price FROM products WHERE id = %s """,(item['product_id'],))price=cursor.fetchone()['price']cursor.execute(""" INSERT INTO order_items (order_id, product_id, quantity, unit_price) VALUES (%s, %s, %s, %s) """,(order_id,item['product_id'],item['quantity'],price))# 5. 记录订单日志cursor.execute(""" INSERT INTO order_logs (order_id, action, details) VALUES (%s, 'CREATED', %s) """,(order_id,f"订单创建，总金额:{total_amount}"))result.update({"success":True,"order_id":order_id,"message":"订单创建成功"})returnresultexceptValueErrorasve:# 业务逻辑错误result["message"]=str(ve)result["errors"].append(str(ve))returnresultexceptpymysql.Errorasdbe:# 数据库错误logger.error(f"数据库错误:{dbe}")result["message"]="系统错误，请稍后重试"result["errors"].append(str(dbe))returnresultexceptExceptionase:# 其他未知错误logger.error(f"未知错误:{e}")result["message"]="系统异常，请联系客服"result["errors"].append(str(e))returnresult

七、学习总结与避坑指南

7.1 关键要点回顾

1. 始终使用事务：对于多个相关操作，一定要放在事务中
2. 及时提交或回滚：避免长事务占用连接资源
3. 合理选择隔离级别：默认REPEATABLE READ适合大多数场景
4. 使用上下文管理器：确保资源正确释放
5. 处理死锁：实现重试机制，统一锁获取顺序

7.2 常见坑点及解决方案

坑点1：忘记提交或回滚

# ❌ 错误：忘记提交conn.autocommit(False)cursor.execute("UPDATE ...")# 忘记 conn.commit() 或 conn.rollback()# 连接关闭时，MySQL会自动回滚，但可能持有锁时间过长# ✅ 正确：使用with语句或try-finallywithtransaction_context(conn)ascursor:cursor.execute("UPDATE ...")

坑点2：在事务中执行DDL语句

# ❌ 错误：事务中执行CREATE TABLEconn.begin()cursor.execute("UPDATE accounts ...")cursor.execute("CREATE TABLE temp ...")# DDL语句会隐式提交！cursor.execute("INSERT INTO temp ...")# 这行不在事务中了！# ✅ 正确：分开执行cursor.execute("CREATE TABLE temp ...")# 先执行DDLconn.begin()# 再开始事务cursor.execute("UPDATE accounts ...")cursor.execute("INSERT INTO temp ...")conn.commit()

坑点3：未处理连接异常

# ❌ 错误：没有连接健康检查deftransfer_money():cursor=conn.cursor()# 如果连接已断开，这里会报错# ✅ 正确：添加连接检查deftransfer_money():ifnotconn.open:conn.ping(reconnect=True)# 尝试重连cursor=conn.cursor()

学习交流与进阶

恭喜你！现在你已经掌握了Python MySQL事务管理的核心技能。事务是数据库编程中最重要也最容易出错的部分，多练习才能真正掌握。

欢迎在评论区分享：

• 你在项目中遇到过哪些事务相关的问题？
• 文中的转账示例你运行成功了吗？
• 对于死锁处理，你有什么实战经验？

我会挑选典型问题详细解答。记住，数据库学习最好的方式就是动手实践！

推荐学习资源：

1. MySQL官方文档 - 事务 - 最权威的事务原理
2. pymysql事务示例 - 官方代码参考
3. 《高性能MySQL》 - 深入理解MySQL内部机制

下篇预告：
下一篇将分享《Python MySQL连接池实战》，教你30分钟搞定DBUtils PooledDB，告别连接超时！

最后的小建议：我刚开始学事务时，总想着"这个简单，理解了"。结果第一次上线就遇到死锁。后来我养成了习惯：每个事务代码写完，都要问自己三个问题：1. 异常时能正确回滚吗？2. 会死锁吗？3. 持有锁的时间长吗？这三个问题能帮你避开80%的坑。

现在，打开你的IDE，把今天的示例代码跑起来吧！遇到问题随时回来讨论。