如何在Django+MySQL 5.5中实现SELECT FOR UPDATE的锁优化（SKIP LOCKED等）

嘿，这个问题我之前维护老版本MySQL的Django项目时也碰到过——MySQL 5.5确实不支持SKIP LOCKED和NOWAIT这两个8.0才引入的原生特性，但咱们还是有办法实现类似效果或者优化锁超时的，下面给你拆解几个适配5.5的方案：

一、模拟“跳过已锁定行”（类似SKIP LOCKED） #

因为MySQL 5.5没有原生的跳过锁行功能，我们可以通过逐个尝试加锁+捕捉超时异常的方式变通实现：

from django.db import transaction, OperationalError

def process_unlocked_records():
    # 先筛选出所有需要处理的候选行主键
    candidate_ids = MyModel.objects.filter(status="pending").values_list("id", flat=True)
    processed_ids = []

    for pk in candidate_ids:
        try:
            # 每个主键单独开启原子事务，尝试加锁
            with transaction.atomic():
                locked_obj = MyModel.objects.select_for_update().get(id=pk)
                # 这里写你的业务处理逻辑
                locked_obj.status = "processed"
                locked_obj.save()
                processed_ids.append(pk)
        except OperationalError as e:
            # 捕捉锁等待超时异常，跳过当前行
            if "Lock wait timeout exceeded" in str(e):
                continue
            # 其他数据库操作错误正常抛出
            else:
                raise
    return processed_ids

这个思路的核心是：把批量加锁拆成单个行的加锁尝试，遇到被锁定的行就直接跳过。虽然牺牲了一点批量操作的原子性，但能满足“跳过锁行”的需求，适合对一致性要求不是极端严格的场景（比如任务队列消费）。

二、实现“立即返回”效果（类似NOWAIT） #

NOWAIT的核心是无法立即获取锁就不等待，直接返回。在MySQL 5.5里，我们可以临时把当前会话的锁等待超时时间设为1秒（最小允许值），这样一旦遇到锁就会立刻触发超时，我们捕捉后返回空结果：

from django.db import connection, transaction, OperationalError

def get_locked_records_nowait():
    try:
        with transaction.atomic():
            # 临时修改当前会话的锁等待超时为1秒
            with connection.cursor() as cursor:
                cursor.execute("SET SESSION innodb_lock_wait_timeout = 1;")
            # 尝试加锁查询
            locked_objs = list(MyModel.objects.select_for_update().filter(status="pending"))
            return locked_objs
    except OperationalError as e:
        if "Lock wait timeout exceeded" in str(e):
            # 遇到锁超时，直接返回空列表
            return []
        # 其他错误正常抛出
        else:
            raise

这样一来，如果查询的行中有任何一行被锁定，当前查询会在1秒内超时，我们就能立刻返回，达到类似NOWAIT的效果。

三、缩短特定查询的锁等待超时时间 #

如果不需要完全跳过或立即返回，只是想缩短特定查询的锁等待时长，同样可以通过临时修改会话级别的innodb_lock_wait_timeout参数实现：

from django.db import connection, transaction

def query_with_custom_lock_timeout():
    with transaction.atomic():
        # 根据业务需求设置超时时间，比如3秒
        with connection.cursor() as cursor:
            cursor.execute("SET SESSION innodb_lock_wait_timeout = 3;")
        # 执行加锁查询和业务逻辑
        target_objs = MyModel.objects.select_for_update().filter(category="priority")
        for obj in target_objs:
            obj.updated_at = timezone.now()
            obj.save()

这个参数是会话级的，只会影响当前事务内的查询，不会修改全局配置，非常适合针对特定查询做优化。

额外提示 #

如果项目有升级空间，最彻底的解决办法是把MySQL升级到8.0+，同时确保Django版本在2.2以上——此时你可以直接使用Django原生支持的select_for_update(skip_locked=True, nowait=True)参数，完全不需要这些变通方案。

内容的提问来源于stack exchange，提问作者akhil_