如何在Django中实现请求队列，管控物理储物柜授权流程？

需求场景 #

• 用户验证授权全部成功：储物柜pin码替换为通用提示+开柜门（已有实现）
• 授权3分钟超时：自动替换pin码为通用提示
• 授权进行中收到新请求：存入队列，前序请求处理完成后再执行

核心问题解决方案 #

1. 请求队列实现（避免pin码被覆盖） #

每个储物柜对应独立的FIFO队列，同时配置处理中状态标记：

• 新请求进入时先判断储物柜状态，未在处理中则直接执行当前请求，生成对应pin码展示，标记状态为处理中
• 如果当前正在处理中，直接把新请求的用户信息、jwt凭证等参数推入对应储物柜的队列即可
• 队列可直接用Redis缓存的list结构实现，不想搭额外服务也可以用Django ORM建AuthorizationQueue表，字段存储：储物柜ID、用户信息、jwt_token、请求时间、请求状态

2. 3分钟授权超时实现 #

用异步延时任务实现超时自动处理，推荐用Celery：

• 每次取出队列请求开始处理时，提交一个延时3分钟执行的超时处理任务，把任务ID存入缓存/队列表
• 用户3分钟内完成授权：先取消对应延时任务，执行原有开柜、展示通用提示的逻辑，再将储物柜状态设为未处理，检查队列是否有待处理请求，有则取出下一个执行
• 3分钟内未完成授权：延时任务自动触发，执行展示通用提示逻辑，设置储物柜状态为未处理，同样检查队列执行下一个请求

补充说明：不想引入Celery依赖的话，也可以用Django Q或者定时任务轮询AuthorizationQueue表中处理中超过3分钟的请求，批量执行超时逻辑，仅实时性稍差。

参考实现代码 #

from django.core.cache import cache
from your_app.celery import app
from celery.result import AsyncResult

# 授权超时时间 单位秒
AUTH_TIMEOUT = 180

def authorization_view(request, jwt_token):
    # 从jwt解析出对应储物柜ID，逻辑自行实现
    bay_id = get_bay_id_from_jwt(jwt_token)
    processing_key = f"bay:{bay_id}:processing"
    queue_key = f"bay:{bay_id}:queue"
    timeout_task_key = f"bay:{bay_id}:timeout_task_id"

    if request.method == 'POST':
        form = ConfirmationForm(request.POST)
        if form.is_valid():
            current_pin = cache.get(f"bay:{bay_id}:current_pin")
            if current_pin == form.cleaned_data['pin']:
                # 授权成功，先取消超时任务
                timeout_task_id = cache.get(timeout_task_key)
                if timeout_task_id:
                    AsyncResult(timeout_task_id).revoke(terminate=True)
                # 原有业务逻辑
                open_bay(jwt_token=jwt_token, pin=current_pin)
                display_generic_message(jwt_token=jwt_token)
                lock_bay(jwt_token=jwt_token, pin=current_pin)
                # 清理当前请求相关缓存
                cache.delete_many([processing_key, timeout_task_key, f"bay:{bay_id}:current_pin"])
                # 处理队列下一个请求
                process_next_queue(bay_id)
                return render(request, 'static/pages/request-success.html')
    else:
        # 新授权请求进入
        if cache.get(processing_key):
            # 推入队列，返回等待提示页
            cache.rpush(queue_key, jwt_token)
            return render(request, 'static/pages/waiting.html', {"msg": "请等待前序用户完成操作"})
        # 未在处理中，直接执行当前请求
        generate_and_display_pin(bay_id)
        cache.set(processing_key, True, AUTH_TIMEOUT + 10)
        # 提交超时任务
        timeout_task = handle_timeout.apply_async((bay_id,), countdown=AUTH_TIMEOUT)
        cache.set(timeout_task_key, timeout_task.id, AUTH_TIMEOUT + 10)
        form = ConfirmationForm()
    return render(request, 'static/pages/confirmation.html', {'form': form})

# 超时处理异步任务
@app.task
def handle_timeout(bay_id):
    display_generic_message(jwt_token=get_current_jwt(bay_id))
    cache.delete_many([f"bay:{bay_id}:processing", f"bay:{bay_id}:timeout_task_key", f"bay:{bay_id}:current_pin"])
    # 处理队列下一个请求
    process_next_queue(bay_id)

# 取出队列下一个请求执行
def process_next_queue(bay_id):
    queue_key = f"bay:{bay_id}:queue"
    if cache.llen(queue_key) > 0:
        next_jwt = cache.lpop(queue_key)
        next_bay_id = get_bay_id_from_jwt(next_jwt)
        generate_and_display_pin(next_bay_id)
        cache.set(f"bay:{next_bay_id}:processing", True, AUTH_TIMEOUT + 10)
        timeout_task = handle_timeout.apply_async((next_bay_id,), countdown=AUTH_TIMEOUT)
        cache.set(f"bay:{next_bay_id}:timeout_task_id", timeout_task.id, AUTH_TIMEOUT + 10)

内容的提问来源于stack exchange，提问作者S Person D Person