Laravel请求串行处理：避免重复创建资源的实现方案

针对多用户同时发起资源创建请求时的串行化需求，核心是保证同一时间仅处理一个创建任务，同时按时间戳顺序执行。你提到的Socket方案并不适配这类场景（Socket主要用于网络通信，无法直接实现进程/线程级的串行调度），下面重点讲两种可行的落地方案：

方案一：原子锁（Atomic Lock）+ 任务队列 #

利用原子性操作实现进程内的互斥锁，结合有序队列维护请求的时间戳顺序：

实现步骤 #

• 1. 接收用户请求时，为请求打上精确时间戳，加入优先级队列（按时间戳排序，确保先到的请求先处理）
• 2. 任务消费逻辑尝试获取原子锁：
  • 若获取成功，从队列头部取出最早的请求执行资源创建
  • 若获取失败，等待锁释放后再尝试
• 3. 资源创建完成后，主动释放原子锁，触发下一个任务的执行

代码示例（Python 进程内场景） #

import time
from queue import PriorityQueue
from threading import Lock

# 全局有序队列，按时间戳排序
request_queue = PriorityQueue()
# 原子锁，保证同一时间仅一个任务执行
create_lock = Lock()

def handle_user_request(user_id, resource_params):
    # 生成精确时间戳作为队列优先级
    timestamp = time.time()
    request_queue.put((timestamp, user_id, resource_params))
    
    # 自动获取/释放锁的上下文管理器
    with create_lock:
        # 取出队列中最早的请求
        ts, uid, params = request_queue.get()
        # 执行资源创建逻辑
        print(f"Processing request from user {uid} at {ts:.2f}")
        # 模拟资源创建耗时
        time.sleep(2)
        print(f"Resource created for user {uid}")

方案二：分布式队列+单消费者（适用于多实例集群） #

如果系统是多实例部署，进程内锁会失效，这时可以用分布式队列配合单消费者模式实现全局串行：

实现步骤 #

• 1. 每个实例接收请求后，将请求（带时间戳）推入Redis有序集合（ZSet），用时间戳作为score保证顺序
• 2. 单独部署一个消费服务（或指定一个实例作为消费者），定时从ZSet中取出score最小的请求
• 3. 消费前通过Redis的SETNX命令获取分布式锁，确保同一时间只有一个消费进程在执行
• 4. 执行完成后，删除ZSet中的该请求并释放锁

代码示例（Redis 分布式场景） #

import redis
import time

r = redis.Redis(host='localhost', port=6379, db=0)

def submit_request(user_id, resource_params):
    timestamp = time.time()
    # 存入ZSet，时间戳作为score保证顺序
    r.zadd('resource_create_queue', {f"{user_id}:{str(resource_params)}": timestamp})

def consume_requests():
    while True:
        # 获取队列中最早的请求
        request = r.zrange('resource_create_queue', 0, 0, withscores=True)
        if not request:
            time.sleep(1)
            continue
        request_str, ts = request[0]
        # 用SETNX实现分布式锁，过期时间30秒防止死锁
        lock_key = 'resource_create_lock'
        if r.set(lock_key, 'locked', nx=True, ex=30):
            try:
                # 解析请求并执行资源创建
                user_id = request_str.split(':')[0]
                print(f"Processing distributed request from user {user_id} at {ts:.2f}")
                time.sleep(2)
                print(f"Distributed resource created for user {user_id}")
                # 删除已处理的请求
                r.zrem('resource_create_queue', request_str)
            finally:
                # 释放锁
                r.delete(lock_key)
        else:
            time.sleep(0.5)

方案选型建议 #

• 单实例部署：优先用方案一，实现简单，性能损耗低
• 多实例集群：必须用方案二，保证分布式环境下的全局串行
• Socket方案不适用：Socket是通信协议，无法直接实现任务的串行调度和互斥控制，属于方向错误

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