FastAPI请求队列实现：GPU图像任务串行处理方案咨询

一、Celery + RabbitMQ 完全适配你的场景 #

这是分布式场景下的标准解决方案，完美匹配你的需求：

• 核心思路是给GPU处理任务的队列配置并发数为1，让Worker串行执行任务，前一个任务完成后才会取下一个。
• 具体操作步骤：
  1. 定义图像处理任务：

     from celery import Celery
     
     # 初始化Celery，指定RabbitMQ作为消息中间件
     app = Celery('image_tasks', broker='amqp://guest@localhost//')
     
     @app.task
     def process_image(image_path):
         # 这里写你的GPU密集计算逻辑
         # 比如调用PyTorch/TensorFlow处理图像
         result = gpu_image_process(image_path)
         return result
     

  2. 启动Worker时指定并发数为1，并绑定专属队列：

     celery -A image_tasks worker --loglevel=info --concurrency=1 -Q image_process_queue
     

  3. 提交请求时统一发送到该队列：

     # 把图像处理请求加入队列
     process_image.apply_async(args=(image_path,), queue='image_process_queue')
     

• 优势：支持分布式部署，后续如果要扩展多GPU节点，只需给每个节点启动一个单并发Worker即可；同时RabbitMQ的持久化机制能保证任务不会丢失。

二、进程内轻量队列（适合单实例部署） #

如果你的应用是单进程/单实例运行，不需要分布式能力，用内置队列+线程锁就能快速实现，更轻量化：

• 用queue.Queue存储待处理请求，启动单独的工作线程串行消费队列：

  import queue
  import threading
  from your_module import gpu_image_process
  
  request_queue = queue.Queue()
  is_running = True
  
  def gpu_worker():
      while is_running:
          try:
              # 阻塞等待新请求，超时1秒避免死等
              image_data, callback = request_queue.get(timeout=1)
              # 执行GPU处理逻辑
              result = gpu_image_process(image_data)
              # 回调返回处理结果（根据业务需求调整）
              callback(result)
              request_queue.task_done()
          except queue.Empty:
              continue
  
  # 启动工作线程（设为守护线程随主进程退出）
  threading.Thread(target=gpu_worker, daemon=True).start()
  
  # 示例：Flask接收请求的视图函数
  def receive_image(request):
      image_data = request.files['image'].read()
      # 定义结果回调函数
      def handle_result(result):
          # 这里写结果返回或存储逻辑
          pass
      # 将请求放入队列
      request_queue.put((image_data, handle_result))
      return "请求已加入队列，等待处理"
  

• 注意：确保GPU处理逻辑是线程安全的，或者工作线程是唯一访问GPU的线程，避免资源冲突。

三、Redis队列（轻量分布式备选） #

如果觉得Celery太重，也可以用Redis实现简单的任务队列，自行实现串行消费：

• 用Redis的LPUSH添加任务，BRPOP阻塞获取任务（同一时间只有一个Worker能拿到任务）：

  import redis
  from your_module import gpu_image_process
  
  # 初始化Redis连接
  r = redis.Redis(host='localhost', port=6379, db=0)
  QUEUE_KEY = 'image_process_queue'
  
  # 生产者：提交图像处理任务
  def submit_task(image_path):
      r.lpush(QUEUE_KEY, image_path)
  
  # 消费者：串行处理任务
  def redis_worker():
      while True:
          # 阻塞等待任务，直到有新请求进入队列
          _, image_path = r.brpop(QUEUE_KEY)
          image_path = image_path.decode('utf-8')
          # 执行GPU处理
          result = gpu_image_process(image_path)
          # 可选：将处理结果存入Redis供查询接口获取
          r.set(f'result:{image_path}', result)
  

• 只需启动一个消费者进程就能实现串行处理；如果要扩展多GPU节点，给每个节点启动一个消费者即可，保证每个GPU节点串行执行任务。

核心注意事项 #

• 无论采用哪种方案，必须保证同一时间只有一个GPU任务在执行，所以Worker/工作线程的并发数必须设为1。
• 如果之前的错误是由于GPU资源冲突（比如显存不足）或任务逻辑不支持并发（比如模型权重被并发修改），串行处理是最直接有效的解决方式。

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