Python使用threading处理任务时能否动态修改、新增queue队列内容？

我刚接触多线程，了解到它非常适配我的使用场景。我目前有一个初始任务队列，如下的程序方案可以满足我的基础需求：

from Queue import Queue
from threading import Thread

def do_stuff(q):
  while True:
    print q.get()
    q.task_done()

q = Queue(maxsize=0)
num_threads = 10

for i in range(num_threads):
  worker = Thread(target=do_stuff, args=(q,))
  worker.setDaemon(True)
  worker.start()

for x in range(100):
  q.put(x)

q.join()

我查询了大量资料，都没有找到能否在队列运行过程中修改、新增队列任务的相关说明。我的业务流程初始有一批任务，部分任务需要等前置任务完成后才可以运行（这类依赖任务规模可达数千个），因此我希望可以根据之前任务的成功/失败状态，持续向队列中新增任务，同时需要限制并发线程的并发数量上限。

首先明确：你使用的Python标准库Queue是线程安全的，完全支持队列运行过程中动态新增、修改任务，你的需求可以在现有方案基础上修改实现。

核心实现要点 #

• 并发数控制：你现有代码中固定启动指定数量worker线程的逻辑已经实现了并发上限控制，不管队列中堆积多少待执行任务，同一时间最多只有你设置的num_threads个任务同时运行，不需要额外调整。
• 动态新增任务：你可以直接在任务处理函数中，根据当前任务的执行结果（成功/失败），调用q.put()方法将后续依赖任务塞入队列即可，多线程环境下调用put方法不会出现竞态问题。
• 依赖调度适配：如果任务有前置依赖，可以将任务封装为包含任务ID、依赖任务ID列表、任务内容的结构体，额外维护一个全局的已完成任务ID集合（读写时加锁保护），worker每次拿到任务先判断依赖是否全部完成，未完成的话可以将任务重新塞回队列末尾等待重试即可。

改进后可运行示例代码 #

from Queue import Queue
from threading import Thread, Lock

# 存储已完成的任务ID，读写操作需加锁保证线程安全
finished_task_ids = set()
lock = Lock()

def do_stuff(q):
    while True:
        task = q.get()
        task_id = task["id"]
        depends_on = task["depends_on"]
        
        # 校验依赖是否全部完成
        lock.acquire()
        dependency_ready = all(dep_id in finished_task_ids for dep_id in depends_on)
        lock.release()
        
        if not dependency_ready:
            # 依赖未满足，将任务放回队列等待下次调度
            q.put(task)
            q.task_done()
            continue
        
        try:
            # 此处替换为你的实际业务处理逻辑
            print(f"执行任务{task_id}, 内容: {task['content']}")
            # 模拟任务执行成功状态，可根据实际执行结果修改
            exec_success = True
            
            if exec_success:
                # 任务执行成功，标记为已完成
                lock.acquire()
                finished_task_ids.add(task_id)
                lock.release()
                
                # 执行成功后新增依赖当前任务的后续任务，可根据业务逻辑调整
                for sub_task_id in [f"{task_id}_sub_1", f"{task_id}_sub_2"]:
                    q.put({
                        "id": sub_task_id,
                        "depends_on": [task_id],
                        "content": f"依赖任务{task_id}的子任务"
                    })
        except Exception as e:
            # 任务执行失败的自定义处理逻辑
            print(f"任务{task_id}执行失败: {str(e)}")
        
        q.task_done()

if __name__ == "__main__":
    q = Queue(maxsize=0)
    # 并发上限，按需调整即可
    num_threads = 10

    for i in range(num_threads):
        worker = Thread(target=do_stuff, args=(q,))
        worker.setDaemon(True)
        worker.start()

    # 初始导入无依赖的根任务
    for x in range(10):
        q.put({
            "id": f"root_{x}",
            "depends_on": [],
            "content": f"初始根任务{x}"
        })

    q.join()

补充优化建议 #

如果依赖任务量级达到数万级别，可以单独维护一个等待队列存储未满足依赖的任务，避免工作队列中塞满待重试的任务拖慢调度效率，每隔固定时间将满足依赖的任务批量移入工作队列即可。如果需要支持修改未执行的任务，可以额外维护一个作废任务ID集合，worker拿到任务先判断是否作废，作废的话直接跳过执行即可。

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