Celery同步队列需求：用户专属动态队列任务串行执行

嘿，针对你开发的这款任务队列应用需求，我来分享一套实用的实现方案，完美满足单用户任务串行执行、多用户任务并行处理的核心要求：

核心逻辑其实很清晰：给每个用户分配一个专属的任务队列，同时为每个队列绑定一个独立的执行单元（线程/协程）。这样一来：

• 单个用户的任务会被放进自己的队列里，执行单元会按顺序逐个取出执行，完成一个再取下一个，自然实现串行；
• 不同用户的执行单元是相互独立的，它们可以同时运行，实现多用户任务的并行处理。

1. 用户队列管理 #

用一个全局的映射结构（比如字典）来关联用户ID和对应的任务队列，确保每个用户的任务都能精准进入自己的专属队列。比如在Python里可以用collections.deque来实现高效的队列操作。

2. 任务执行器 #

为每个用户的队列启动一个长期运行的执行线程（或协程，根据你的开发语言和场景选择）。这个执行器会循环检查自己负责的队列：

• 如果队列里有任务，就取出第一个开始执行；
• 任务执行完成后，再检查队列是否有新任务，重复这个流程；
• 如果队列为空，就短暂休眠避免空循环浪费资源，之后继续监听。

3. 动态初始化执行器 #

当某个用户第一次添加任务时，自动为其创建队列并启动对应的执行线程，不需要提前预创建所有用户的执行器，节省资源。如果某个用户的队列长期为空，还可以考虑回收执行线程（可选优化项）。

下面是一个简单的可运行示例，用线程实现核心逻辑：

import threading
from collections import deque
import time
import random

# 全局存储用户ID到任务队列的映射
user_task_queues = {}
# 线程锁：避免多线程同时修改队列导致的并发问题
queue_lock = threading.Lock()

def task_executor(user_id):
    """每个用户专属的任务执行线程"""
    while True:
        current_task = None
        # 加锁操作队列，保证线程安全
        with queue_lock:
            if user_id in user_task_queues and len(user_task_queues[user_id]) > 0:
                current_task = user_task_queues[user_id].popleft()
        
        if current_task:
            print(f"[用户{user_id}] 开始执行任务: {current_task['name']}")
            # 模拟任务执行耗时（替换成你的实际任务逻辑）
            time.sleep(random.randint(1, 3))
            print(f"[用户{user_id}] 完成任务: {current_task['name']}")
        else:
            # 队列为空时短暂休眠，减少CPU占用
            time.sleep(0.5)

def add_user_task(user_id, task_info):
    """对外暴露的添加任务接口"""
    with queue_lock:
        if user_id not in user_task_queues:
            # 首次添加任务，创建队列并启动执行线程
            user_task_queues[user_id] = deque()
            # 设置为守护线程，主进程退出时自动结束
            threading.Thread(target=task_executor, args=(user_id,), daemon=True).start()
        user_task_queues[user_id].append(task_info)
    print(f"已将任务{task_info['name']}加入用户{user_id}的队列")

# 测试场景
if __name__ == "__main__":
    # 用户1添加3个串行任务
    add_user_task("User1", {"name": "Task1"})
    add_user_task("User1", {"name": "Task2"})
    add_user_task("User1", {"name": "Task3"})
    
    # 用户2添加2个串行任务，会和User1的任务并行执行
    add_user_task("User2", {"name": "TaskA"})
    add_user_task("User2", {"name": "TaskB"})
    
    # 等待所有任务执行完成（生产环境可改用事件通知等更优雅的方式）
    time.sleep(10)

运行这个示例你会看到：User1的Task1、Task2、Task3是按顺序执行的，而User2的TaskA会和User1的Task1同时运行，完全符合你的需求。

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