1. 从一次数据错乱说起为什么ORM里还需要关心锁和事务刚接触Django那会儿我总觉得ORM已经把数据库操作封装得足够优雅了Model.objects.get()、save()、filter()这些API用起来行云流水谁还会去操心底层的锁和事务直到有一次做一个库存扣减的功能测试环境跑得好好的一上生产就出问题——两个用户同时下单库存明明只剩1件结果两个订单都扣减成功了库存变成了-1。排查了半天最后定位到问题根源并发场景下没有正确处理事务和行级锁。这件事让我彻底明白一个道理ORM再方便它也只是帮你生成SQL的工具数据库层面的并发控制、事务隔离、锁机制该懂的还是得懂。Django的ORM提供了transaction.atomic()、select_for_update()、F()表达式等工具来应对这些问题但很多人只是知道有这么个东西具体什么场景该用哪个、怎么用才不出问题心里其实没底。这篇内容就是把我自己在实际项目里踩过的坑、总结的经验完整梳理一遍。从Django ORM的事务机制讲起到select_for_update的行级锁实战再到并发场景下的常见误区和排查技巧都会涉及。适合已经用过Django ORM做基本增删改查、但对事务和锁机制还不够熟悉的开发者。如果你正在做电商、金融、票务这类对数据一致性要求高的系统这些内容应该能帮你少走不少弯路。2. Django ORM事务机制全拆解2.1 默认自动提交模式下的隐患Django默认运行在自动提交autocommit模式下。什么意思呢就是你每执行一条写操作SQLDjango会立刻把它提交到数据库不会等你显式调用commit。比如你写了这么一段代码order Order.objects.create(useruser, amount100) balance Balance.objects.get(useruser) balance.amount - 100 balance.save()这三步操作在自动提交模式下是三条独立的事务。如果第一步创建订单成功了第二步扣减余额的时候抛了异常那订单已经写进数据库了但余额没扣——数据就不一致了。这就是为什么需要显式地把多个操作包在一个事务里。Django提供了两种方式来控制事务边界一种是全局配置ATOMIC_REQUESTS另一种是用transaction.atomic()装饰器或上下文管理器。两种方式各有适用场景后面会详细对比。2.2 atomic的三种用法与底层原理transaction.atomic()是Django中最核心的事务控制工具它有三种常见用法。第一种作为装饰器用在视图函数上from django.db import transaction transaction.atomic def create_order(request): # 这里面的所有数据库操作都在一个事务里 order Order.objects.create(...) Balance.objects.filter(userrequest.user).update(...) return JsonResponse({status: ok})第二种作为上下文管理器def create_order(request): with transaction.atomic(): order Order.objects.create(...) balance Balance.objects.select_for_update().get(userrequest.user) balance.amount - order.amount balance.save() # 出了with块事务提交 return JsonResponse({status: ok})第三种作为装饰器用在非视图函数上比如Service层的方法transaction.atomic def transfer_funds(from_account, to_account, amount): # 转账逻辑 ...底层原理其实不复杂。transaction.atomic()进入的时候会做两件事一是关闭自动提交模式如果之前是开启的二是记录一个保存点savepoint。如果块内的代码正常执行完毕Django会提交事务如果抛了异常就回滚到保存点。这里有个关键细节atomic块可以嵌套。嵌套的时候外层atomic负责真正的提交和回滚内层atomic只是设置一个保存点。如果内层抛异常且被捕获了只会回滚到内层的保存点外层事务可以继续。这个嵌套特性在实际项目中非常有用。比如你有一个大的事务包裹了整个请求处理流程但其中某个子步骤失败了你不想让整个事务回滚就可以在内层用atomic包一下然后捕获异常做降级处理。2.3 ATOMIC_REQUESTS全局配置的利与弊在settings.py的DATABASES配置里可以设置ATOMIC_REQUESTS: TrueDATABASES { default: { ENGINE: django.db.backends.mysql, NAME: mydb, ... ATOMIC_REQUESTS: True, } }开了这个配置之后每个请求都会被自动包在一个事务里。请求开始时开启事务正常返回时提交抛异常时回滚。听起来很省事对吧但它有几个坑你得知道。第一个坑它会把只读请求也包进事务。一个简单的GET请求本来不需要事务现在也开了一个。在高并发场景下这会增加数据库的连接持有时间和锁竞争。第二个坑中间件的执行顺序会影响事务边界。如果你在ATOMIC_REQUESTS开启的情况下又在视图里手动用了transaction.atomic()那就变成了嵌套事务。虽然Django能处理但调试起来会让人头疼。第三个坑流式响应和长连接场景下不适用。如果你的视图返回的是StreamingHttpResponse事务会在响应开始流式输出时还持有可能导致连接池耗尽。我个人的建议是不要全局开ATOMIC_REQUESTS而是在需要事务的视图或Service方法上显式使用atomic。这样事务边界清晰性能也更好控制。当然如果你的项目规模不大、并发不高全局开启也不是不行但一定要知道它的代价。2.4 事务与异常处理的配合要点用atomic的时候异常处理有个非常容易踩的坑在atomic块内部捕获了数据库异常但没有重新抛出会导致事务状态混乱。with transaction.atomic(): try: obj MyModel.objects.get(pk1) obj.field new_value obj.save() except Exception as e: logger.error(f出错了: {e}) # 这里吞掉了异常但事务可能已经处于broken状态 # 继续执行其他数据库操作 OtherModel.objects.create(...) # 这行可能报TransactionManagementErrorDjango在检测到事务中有异常发生后如果你不把异常抛出atomic块后续的数据库操作会抛出TransactionManagementError。正确的做法是要么让异常自然传播出去让atomic回滚要么在内层用嵌套atomic把可能出错的操作用保存点包起来。with transaction.atomic(): try: with transaction.atomic(): # 内层保存点 obj MyModel.objects.get(pk1) obj.field new_value obj.save() except Exception as e: logger.error(f出错了: {e}) # 外层事务可以继续 OtherModel.objects.create(...)这个模式在实际项目中非常实用尤其是批量处理场景——某一条记录处理失败不应该影响整批。3. select_for_update行级锁实战解析3.1 什么时候必须用行级锁先看一个典型的并发问题场景。假设有一个账户余额表两个请求同时要扣减余额# 请求A和请求B同时执行 balance Balance.objects.get(user_id1) # 两边都读到100 balance.amount - 30 # 两边都算成70 balance.save() # 两边都写入70期望结果是40实际结果是70。这就是经典的**丢失更新Lost Update**问题。原因在于读取和写入之间没有锁保护两个事务读到了同一个旧值。解决这个问题有三种思路一是用F()表达式做原子更新二是用select_for_update()加行级锁三是用乐观锁版本号。F()表达式适合简单的自增自减场景from django.db.models import F Balance.objects.filter(user_id1).update(amountF(amount) - 30)这条SQL在数据库层面是原子的不会有并发问题。但如果你需要先读取余额做业务判断比如判断余额是否充足然后再扣减F()表达式就不够用了必须用select_for_update()。3.2 select_for_update的参数详解select_for_update()的基本用法from django.db import transaction with transaction.atomic(): balance Balance.objects.select_for_update().get(user_id1) if balance.amount 30: balance.amount - 30 balance.save() else: raise InsufficientBalanceError(余额不足)它生成的SQL大致是SELECT ... FOR UPDATE会在选中的行上加排他锁。其他事务如果也想对这行加锁就会被阻塞直到当前事务提交或回滚。select_for_update()支持几个重要参数参数说明适用场景nowaitTrue如果行已被锁定立即抛异常不等待避免长时间阻塞适合快速失败场景skip_lockedTrue跳过已被锁定的行任务队列消费场景of(self,)指定锁定的表涉及join查询时精确控制no_keyTrue只加表级锁不加行级锁PostgreSQL特有极少使用nowait和skip_locked在实际项目中很有价值。比如做一个异步任务队列多个worker同时从任务表里取任务用skip_lockedTrue就能避免worker之间互相等待with transaction.atomic(): task Task.objects.select_for_update(skip_lockedTrue).filter( statuspending ).first() if task: task.status processing task.save()这样每个worker拿到的是不同的任务不会出现一个任务被多个worker抢到的情况。3.3 锁的粒度与数据库差异select_for_update()的行为在不同数据库下有关键差异这一点很多人容易忽略。MySQLInnoDB默认加的是行级排他锁。但要注意如果查询条件没有走索引InnoDB可能会退化成表级锁。比如Balance.objects.select_for_update().filter(name张三)如果name字段没有索引MySQL会锁住整张表。这个坑非常隐蔽在高并发下会导致严重的性能问题。PostgreSQL行级锁行为更精确支持nowait和skip_locked更完善。但PostgreSQL有个特点——如果SELECT FOR UPDATE没有匹配到任何行不会加锁这在某些场景下需要注意。SQLiteSQLite不支持SELECT FOR UPDATE语法。Django在SQLite上调用select_for_update()会直接忽略锁不会报错但也不会有锁效果。所以如果你的开发环境用SQLite、生产环境用MySQL并发测试一定要在MySQL上做否则测不出问题。注意使用select_for_update()必须在事务块内否则Django会抛出TransactionManagementError。这是因为锁只在事务期间有效事务一提交锁就释放了。3.4 死锁的成因与规避策略用了行级锁之后死锁就成了绕不开的话题。死锁的经典场景是两个事务以不同顺序锁定多行# 事务A with transaction.atomic(): acc1 Account.objects.select_for_update().get(id1) acc2 Account.objects.select_for_update().get(id2) # 转账操作 # 事务B同时执行 with transaction.atomic(): acc2 Account.objects.select_for_update().get(id2) # 先锁2 acc1 Account.objects.select_for_update().get(id1) # 再锁1事务A持有id1的锁等id2事务B持有id2的锁等id1互相等待死锁形成。数据库通常会检测到死锁并回滚其中一个事务但你的代码需要处理这个异常。规避死锁的核心原则是统一加锁顺序。比如所有转账操作都按账户ID从小到大加锁with transaction.atomic(): ids sorted([from_id, to_id]) accounts Account.objects.select_for_update().filter( id__inids ).order_by(id) # 按固定顺序处理另一个策略是缩短锁持有时间。锁的持有时间等于事务持续时间事务里不要做网络请求、文件IO这些耗时操作尽快提交。4. 并发场景下的完整实操方案4.1 库存扣减的三种实现对比回到开头提到的库存扣减问题我用三种方案来实现对比一下各自的优劣。方案一F表达式原子更新def deduct_stock(product_id, quantity): updated Product.objects.filter( idproduct_id, stock__gtequantity ).update(stockF(stock) - quantity) if updated 0: raise InsufficientStockError(库存不足)这个方案最简洁一条SQL搞定filter里的stock__gtequantity保证了不会超卖。缺点是如果扣减后还需要做其他操作比如记录流水需要额外查询。方案二select_for_update悲观锁transaction.atomic def deduct_stock(product_id, quantity): product Product.objects.select_for_update().get(idproduct_id) if product.stock quantity: raise InsufficientStockError(库存不足) product.stock - quantity product.save() StockLog.objects.create( product_idproduct_id, quantityquantity, typededuct )这个方案适合扣减后需要做多个关联操作的场景所有操作在一个事务里要么全成功要么全回滚。方案三乐观锁版本号def deduct_stock(product_id, quantity, max_retries3): for _ in range(max_retries): product Product.objects.get(idproduct_id) if product.stock quantity: raise InsufficientStockError(库存不足) updated Product.objects.filter( idproduct_id, versionproduct.version ).update( stockproduct.stock - quantity, versionproduct.version 1 ) if updated: return raise ConcurrentConflictError(并发冲突请重试)乐观锁适合冲突不频繁的场景不需要数据库加锁性能更好。但冲突频繁时重试次数会增加反而影响体验。三种方案的选型建议方案适用场景优点缺点F表达式简单原子增减简洁高效无法做复杂业务判断悲观锁复杂关联操作逻辑清晰锁竞争影响性能乐观锁冲突少的场景无锁性能好冲突多时重试开销大4.2 转账功能的完整实现转账是事务和锁的经典应用场景我把完整实现写出来包含参数计算和异常处理。from django.db import transaction from django.db.models import F from decimal import Decimal class InsufficientBalanceError(Exception): pass transaction.atomic def transfer(from_user_id, to_user_id, amount): 转账操作amount为Decimal类型 使用select_for_update锁定两个账户按ID排序避免死锁 if amount Decimal(0): raise ValueError(转账金额必须大于0) # 按ID排序统一加锁顺序 ids sorted([from_user_id, to_user_id]) accounts { acc.user_id: acc for acc in Account.objects.select_for_update().filter( user_id__inids ).order_by(user_id) } from_acc accounts[from_user_id] to_acc accounts[to_user_id] if from_acc.balance amount: raise InsufficientBalanceError( f余额不足: 当前{from_acc.balance}, 需要{amount} ) from_acc.balance - amount to_acc.balance amount from_acc.save(update_fields[balance]) to_acc.save(update_fields[balance]) # 记录流水 TransferLog.objects.create( from_user_idfrom_user_id, to_user_idto_user_id, amountamount, )这里有几个细节值得说明。第一select_for_update()配合filter(id__inids).order_by(user_id)一次性锁定两个账户减少了数据库往返。第二save(update_fields[balance])只更新余额字段减少写入量。第三金额用Decimal而不是float避免浮点精度问题——这在金融场景是铁律。4.3 批量操作中的事务控制批量处理数据的时候事务粒度是个需要仔细考虑的问题。事务太大锁持有时间长回滚代价高事务太小频繁提交影响性能。我一般用分批提交的策略def batch_process(items, batch_size500): for i in range(0, len(items), batch_size): batch items[i:i batch_size] try: with transaction.atomic(): for item in batch: process_item(item) except Exception as e: logger.error(f批次{i//batch_size}处理失败: {e}) # 记录失败批次继续处理下一批 FailedBatch.objects.create( batch_indexi // batch_size, errorstr(e) )每500条一个事务单批失败不影响其他批次。这个batch_size不是拍脑袋定的要考虑几个因素单条记录的处理耗时、数据库的max_allowed_packet配置、以及业务能容忍的数据不一致窗口。我实测下来对于简单的更新操作500到1000是比较稳妥的范围。4.4 异步任务中的事务边界Celery异步任务里用事务要特别小心。因为Celery任务可能在事务提交之前就开始执行了导致读不到最新数据。比如transaction.atomic def create_order_and_notify(user, product): order Order.objects.create(useruser, productproduct) # 错误做法事务还没提交就发异步任务 send_notification.delay(order.id) # worker可能查不到这个order return order正确做法是用transaction.on_commit()transaction.atomic def create_order_and_notify(user, product): order Order.objects.create(useruser, productproduct) transaction.on_commit( lambda: send_notification.delay(order.id) ) return orderon_commit注册的回调会在事务成功提交后才执行。这个机制在涉及缓存更新、消息发送、异步任务触发时都应该用上。5. 常见问题排查与避坑指南5.1 事务不生效的几种典型情况情况一在atomic块外调用了select_for_update。这个前面提过Django会直接报错。但还有一种隐蔽的情况——你在一个被transaction.atomic装饰的函数里调用了另一个函数那个函数里用了select_for_update看起来是在事务里但如果装饰器没生效比如装饰的是类方法但忘了加method_decorator就会报错。情况二数据库引擎不支持事务。MySQL的MyISAM引擎不支持事务如果你建表的时候没指定InnoDBatomic块里的操作实际上没有事务保护。Django默认建表用的是InnoDB但如果是接手的老项目一定要确认一下。情况三多数据库场景下事务没覆盖到。如果你的项目配置了多个数据库transaction.atomic()默认只作用于default数据库。要操作其他数据库的事务需要指定using参数with transaction.atomic(usingreplica): ...5.2 锁等待超时的排查思路生产环境遇到锁等待超时排查步骤大致是这样第一步确认是哪个事务持有了锁。MySQL可以用SHOW ENGINE INNODB STATUS查看当前锁等待情况或者查information_schema.INNODB_TRX和INNODB_LOCK_WAITS表。第二步看持有锁的事务在执行什么SQL。如果是一个长事务可能是代码里有耗时操作没及时提交。第三步检查查询条件是否走了索引。前面说过没走索引的SELECT FOR UPDATE可能锁全表这是最常见的锁等待原因。第四步考虑是否需要调整innodb_lock_wait_timeout参数。默认是50秒对于Web应用来说太长了一般调到5到10秒比较合适让请求快速失败而不是一直挂着。5.3 常见问题速查表问题现象可能原因排查方法解决方案TransactionManagementErroratomic块内吞了异常检查try-except位置用嵌套atomic包住可能出错的代码数据更新丢失没用锁或F表达式检查并发写逻辑加select_for_update或F表达式锁等待超时长事务或全表锁查INNODB_TRX缩短事务、加索引、调超时参数死锁加锁顺序不一致查死锁日志统一加锁顺序异步任务读不到数据事务未提交就触发检查任务触发时机用transaction.on_commitSQLite下锁无效SQLite不支持行锁确认数据库类型并发测试用MySQL/PostgreSQL5.4 几个我踩过的坑坑一在atomic块里调用了外部API。有一次写了个支付回调处理在事务里调用了第三方支付接口查询订单状态结果第三方接口响应慢事务持有了好几秒把连接池占满了。教训是事务里绝对不要做网络IO先把数据查出来提交事务再去调外部接口。坑二select_for_update配合values()使用。select_for_update()必须配合get()、filter()、first()这些返回模型实例的方法使用。如果你用values()或values_list()锁不会生效因为Django不知道怎么把锁映射到模型实例上。坑三以为atomic能解决所有并发问题。atomic只保证事务的原子性不解决并发读写冲突。两个事务同时读同一行数据atomic不会阻止它们必须靠锁或乐观锁来协调。坑四在测试用例里测并发。Django的TestCase默认把每个测试包在事务里测试结束回滚。这意味着你在测试里用select_for_update锁的行为和真实环境不一样。要测并发得用TransactionTestCase而且最好在真实数据库上测别用SQLite。6. 一些实战中的经验体会关于事务粒度我的经验是能小则小。一个事务里只放必须原子化的数据库操作其他逻辑参数校验、日志记录、消息组装都放到事务外面。事务越小锁持有时间越短并发性能越好。关于锁的选择优先用F表达式其次乐观锁最后才用悲观锁。F表达式最简单高效乐观锁在冲突少的场景下性能好悲观锁虽然逻辑直观但锁竞争代价大。只有在业务逻辑复杂、必须读取后判断再写入的场景才用select_for_update。关于监控生产环境一定要监控慢查询和锁等待。Django的connection.queries可以看SQL执行情况但生产环境更推荐用数据库自带的慢查询日志。锁等待的监控可以用performance_schema或者云数据库的监控面板。早发现早优化别等出事了再查。最后分享一个调试小技巧在开发环境可以用django.db.connection.queries打印出ORM实际执行的SQL配合transaction.atomic的嵌套能清楚看到每个操作在哪个事务里、加了什么锁。这个习惯帮我定位过好几次隐蔽的并发问题。