Django+Gunicorn多Worker共享锁实现：避免重复执行视图循环

你用multiprocessing.Lock()无效的核心原因是：Gunicorn的每个Worker都是独立的Python进程，每个进程启动时会重新加载你的代码，所以每个Worker里的lock都是各自独立的实例，完全无法跨进程同步。

下面是几种可行的跨Worker同步方案：

方案一：Redis分布式锁 #

Redis的原子操作可以实现可靠的跨进程/跨机器同步锁，是这类场景的常用解决方案。

步骤： #

1. 安装Redis依赖库：

pip install redis

2. 视图中使用Redis锁：

import redis
import time
from django.conf import settings
from django.http import HttpResponse

# 全局初始化Redis连接（避免每次请求新建连接）
redis_client = redis.Redis(host=settings.REDIS_HOST, port=settings.REDIS_PORT, db=0)

def loop_func(request):
    # 定义唯一的锁标识
    lock_key = "loop_func_exclusive_lock"
    # 设置锁超时时间（必须长于循环执行时长，防止进程崩溃导致死锁）
    with redis_client.lock(lock_key, timeout=120):
        # 执行你的循环逻辑
        for i in range(0, 100):
            print("did work ", i)
            time.sleep(1)
    return HttpResponse("Loop completed")

注意： #

• 超时时间要比循环实际执行时间长，避免锁提前释放导致并发问题。
• 如果循环执行时长不确定，可以添加锁续约逻辑（比如用后台线程定时延长超时）。

方案二：文件锁（单机器部署适用） #

利用操作系统的文件锁机制，多个进程通过同一个文件的锁实现同步，无需额外服务依赖。

代码示例（Linux/Unix）： #

import fcntl
import time
from django.http import HttpResponse

# 锁文件路径，确保所有Worker都能读写
LOCK_FILE_PATH = "/tmp/loop_func_lock"

def loop_func(request):
    lock_file = open(LOCK_FILE_PATH, 'w')
    try:
        # 加排他锁，阻塞直到获取锁
        fcntl.flock(lock_file, fcntl.LOCK_EX)
        # 执行循环逻辑
        for i in range(0, 100):
            print("did work ", i)
            time.sleep(1)
    finally:
        # 释放锁并关闭文件
        fcntl.flock(lock_file, fcntl.LOCK_UN)
        lock_file.close()
    return HttpResponse("Loop completed")

注意： #

• Windows系统需使用win32file模块的LockFileEx/UnlockFileEx实现，逻辑类似但API不同。
• 锁文件要放在所有Worker都有权限访问的路径下。

方案三：数据库行锁（利用现有数据库） #

如果项目已使用数据库，可以直接用数据库的行锁实现同步，无需额外依赖。

步骤： #

1. 创建锁表（Django模型）：

from django.db import models

class TaskLock(models.Model):
    lock_name = models.CharField(max_length=100, unique=True)
    is_locked = models.BooleanField(default=False)

    class Meta:
        db_table = "task_locks"

执行迁移创建表：

python manage.py makemigrations && python manage.py migrate

2. 视图中使用数据库锁：

import time
from django.db import transaction
from django.http import HttpResponse
from .models import TaskLock

def loop_func(request):
    lock_name = "loop_func_lock"
    # 事务内使用select_for_update加排他锁，阻塞直到获取锁
    with transaction.atomic():
        # 获取或创建锁记录
        lock, _ = TaskLock.objects.get_or_create(lock_name=lock_name)
        # 加行锁，其他进程会阻塞到当前事务提交
        lock = TaskLock.objects.select_for_update().get(lock_name=lock_name)
        # 执行循环逻辑
        for i in range(0, 100):
            print("did work ", i)
            time.sleep(1)
        # 事务提交后自动释放锁
    return HttpResponse("Loop completed")

注意： #

• 必须在事务上下文内使用select_for_update，否则锁不会生效。
• 部分数据库需要开启事务支持才能使用行锁。

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