Redis任务资源释放及断点续跑方案咨询

针对你当前无限轮询导致资源浪费、进程无法复用的问题，以下是几种实用的优化方案：

方案1：Redis发布/订阅（Pub/Sub）模式 #

核心逻辑：放弃主动轮询，改为被动接收任务完成通知，实现进程复用。

• 实现步骤：
  1. 任务启动时生成唯一task_id，等待进程订阅Redis的task:finished频道（也可为每个任务创建专属频道）。
  2. 任务执行完成后，在任务逻辑中向对应频道发布包含task_id和结果的消息。
  3. 等待进程收到消息后，匹配目标task_id，通过Socket向前端推送结果，之后保持监听状态，继续处理下一个任务的通知。
• 伪代码示例：

  import redis
  import json
  
  r = redis.Redis()
  pubsub = r.pubsub()
  pubsub.subscribe('task:finished')
  
  # 单进程持续监听，可处理多个任务通知
  for message in pubsub.listen():
      if message['type'] == 'message':
          task_data = json.loads(message['data'])
          target_task_id = task_data['task_id']
          # 匹配当前需要处理的任务，推送结果到前端
          send_socket_message(target_task_id, task_data['result'])
  

方案2：Redis阻塞队列（BLPOP/BRPOP） #

核心逻辑：利用Redis的阻塞命令让等待进程进入休眠状态，避免空轮询占用CPU，同时支持进程复用。

• 实现步骤：
  1. 任务启动时创建专属阻塞队列，比如task:queue:{task_id}。
  2. 等待进程调用BRPOP task:queue:{task_id} 0（0表示无限阻塞），此时进程释放CPU资源进入休眠。
  3. 任务完成后，将结果LPUSH到对应队列，等待进程被唤醒，取出结果推送后，继续处理下一个任务的阻塞请求。
• 伪代码示例：

  import redis
  
  r = redis.Redis()
  
  def handle_task_wait(task_id):
      # 阻塞等待队列数据，无数据时休眠不占用CPU
      _, task_result = r.brpop(f'task:queue:{task_id}', timeout=0)
      # 推送结果到前端
      send_socket_message(task_id, task_result)
  
  # 复用进程循环处理多个等待任务
  while True:
      next_task_id = get_pending_task_id()  # 从任务池获取下一个需要等待的任务ID
      handle_task_wait(next_task_id)
  

方案3：任务回调机制（自定义/框架原生） #

如果你的任务系统基于Celery等常用Redis任务框架，可直接利用回调机制跳过等待环节：

• 框架原生方案：定义任务成功回调函数，在回调中直接通过Socket向前端推送结果，无需额外等待进程，资源由框架自动管理，进程可复用处理其他任务。
• 自定义任务系统：在任务执行结束的最后一步，主动触发Socket推送逻辑，彻底省去中间等待步骤：

  def execute_task(task_id):
      # 任务执行逻辑
      task_result = run_redis_job()
      # 任务完成后直接推送结果
      send_socket_message(task_id, task_result)
  

方案对比 #

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