select_for_update() 行锁实验

SQL 与前提

实测 · 本机执行输出

不在 atomic 里 → TransactionManagementError: select_for_update cannot be used outside of a transaction.
atomic 里的 SQL: lab_task"."stock" FROM "lab_task" WHERE "lab_task"."id" = 1 FOR UPDATE

行锁的生命周期就是事务的生命周期:事务提交或回滚时释放。

抢同一行

实测 · 本机执行输出

持锁事务持有 1.0 s 期间:
  nowait=True    → OperationalError: could not obtain lock on row in relation "lab_task"
  skip_locked=True → skip_locked 返回 0 行
  普通 select_for_update → 拿到锁,等了 1.06 s (释放锁之后才返回)

三种策略:等(默认)、失败(nowait)、跳过(skip_locked)。skip_locked 意味着你看到的是「当前没被别人锁住的那部分」,不是完整的集合——适合队列,不适合需要完整视图的场景。

任务队列

实测 · 本机执行输出

各工作线程处理的任务数: [10, 10, 10]  合计 30
每个任务被几个人处理过(done_by 非空): 30 / 30 ;done_by 分布: {'w3': 10, 'w2': 10, 'w1': 10}

典型写法:with atomic(): t = qs.select_for_update(skip_locked=True).first(); 处理; 标记完成。本实验里每个任务处理了 10 ms;处理时间更长、失败时事务回滚会让任务重新变为可领取。

of

实测 · 本机执行输出

of=("self") 的 SQL 尾部: k"."name" = 'a' FOR UPDATE OF "lab_task"

有 select_related / JOIN 时,用 of=('self',) 避免锁住不该锁的行。

分步演示:抢锁

select_for_update

本步的对象

本文用到的工具与边界

本机 macOS 27 / Python 3.14.7 / Django 6.1.1(默认用 SQLite,其它数据库会在文中注明),用 django.test.Client 或真实的本机服务器进程实测;计时用 time.perf_counter,数字每次运行会有小幅抖动。PostgreSQL 16.15(本机临时实例,端口 9422),READ COMMITTED;每个线程独立连接;「等待 1.06 s」是实测的一次数字
没有验证:MySQL / SQLite 对 select_for_update 的支持(SQLite 忽略、MySQL 8 支持 skip_locked)、死锁场景、no_key 选项、on_commit 与锁释放的关系

下一篇 · DJANGO 47

数据库隔离级别与 Django