如何在Python中创建可后台运行任务的线程队列？

嘿，我来帮你搞定这个问题——首先得澄清一个常见误解：asyncio.Queue.put()本身不会直接启动任务，你之前遇到的“放入就启动”，大概率是因为你把已经通过asyncio.create_task()创建好的、正在运行的协程对象塞进去了，而不是待处理的原始数据（比如文件名）。

结合你的需求：定时从数据库拉取未知数量的文件名，持续运行并处理新文件，我给你一套完整的实现方案，刚好匹配你的场景。

核心思路拆解 #

• 生产者协程：按指定间隔轮询数据库，捞取新增的文件名，把这些文件名推入队列。
• 消费者协程：从队列里取文件名，执行你的文件处理逻辑（可以开多个消费者提高并发效率）。
• 主程序：启动生产者和消费者，让事件循环一直跑，直到你手动停止程序。

完整代码示例 #

import asyncio
from datetime import datetime

# ---------------------- 替换成你的实际逻辑 ----------------------
async def fetch_new_files_from_db():
    """模拟从数据库获取新增文件名，实际替换成你的数据库查询代码"""
    print(f"[{datetime.now().strftime('%H:%M:%S')}] 正在查询数据库...")
    # 这里写你的数据库操作：比如查最近N分钟新增的文件记录
    # 下面是模拟返回随机数量的新文件
    import random
    new_files = [f"document_{i}.pdf" for i in range(random.randint(0, 4))]
    return new_files

async def process_file(filename):
    """替换成你的实际文件处理逻辑：比如解析、上传、转码等"""
    print(f"[{datetime.now().strftime('%H:%M:%S')}] 开始处理: {filename}")
    # 模拟处理耗时（比如读取文件、调用API等）
    await asyncio.sleep(1.5)
    print(f"[{datetime.now().strftime('%H:%M:%S')}] 处理完成: {filename}")
# ----------------------------------------------------------------

async def producer(queue, poll_interval):
    """生产者：定时查库，推送新文件到队列"""
    while True:
        new_files = await fetch_new_files_from_db()
        if new_files:
            print(f"[{datetime.now().strftime('%H:%M:%S')}] 发现{len(new_files)}个新文件，加入队列")
            for file in new_files:
                # 这里只放文件名，不是运行中的协程！
                await queue.put(file)
        # 等待指定间隔后再次查询
        await asyncio.sleep(poll_interval)

async def consumer(queue, consumer_id):
    """消费者：从队列取文件并处理"""
    while True:
        # 阻塞等待队列里的新文件
        filename = await queue.get()
        try:
            # 拿到文件名后才启动处理逻辑
            await process_file(filename)
        finally:
            # 标记任务完成，队列用来跟踪未完成任务数（可选但推荐）
            queue.task_done()

async def main():
    # 创建队列，maxsize可选：如果队列满了，生产者会阻塞直到有空闲位置
    task_queue = asyncio.Queue(maxsize=15)
    
    # 启动生产者，设置轮询间隔（比如每10秒查一次库）
    producer_task = asyncio.create_task(producer(task_queue, poll_interval=10))
    
    # 启动3个消费者，根据你的处理能力调整数量
    consumer_tasks = [
        asyncio.create_task(consumer(task_queue, i+1)) 
        for i in range(3)
    ]
    
    # 让程序持续运行，直到按下Ctrl+C中断
    try:
        await asyncio.gather(producer_task, *consumer_tasks)
    except KeyboardInterrupt:
        print("\n收到中断信号，正在优雅退出...")
        # 取消所有任务
        producer_task.cancel()
        for task in consumer_tasks:
            task.cancel()
        # 等待任务完成取消流程
        await asyncio.gather(producer_task, *consumer_tasks, return_exceptions=True)
        print("程序已退出")

if __name__ == "__main__":
    asyncio.run(main())

关键细节解释 #

1. 队列存数据而非运行协程：这里我们只把文件名放入队列，消费者拿到数据后才调用process_file执行处理逻辑——这就彻底解决了“放入队列就启动”的问题。
2. 定时轮询的正确姿势：用while True + asyncio.sleep()实现定时查库，比用loop.call_later()更直观，适合持续运行的场景。
3. 多消费者并发：启动多个消费者协程，可以同时处理多个文件，适合文件处理耗时较长的场景，提高整体效率。
4. 优雅退出：捕获KeyboardInterrupt信号，取消所有任务并等待完成，避免强制退出导致的数据库连接未关闭、文件处理到一半等问题。

如果你之前的错误操作是放入了协程 #

如果你之前是这么写的：

await queue.put(asyncio.create_task(process_file(file)))

那确实会立即启动任务，因为asyncio.create_task()会直接把协程加入事件循环执行。正确的做法是只放文件名（或待处理的参数），由消费者来创建并执行任务。

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