多 Worker 分布式渲染：从 HTTP 轮询到 RocketMQ

方糖猫 方糖猫
- 约 7 分钟读完 - 全文 2498 字

多 Worker 分布式渲染：从 HTTP 轮询到 RocketMQ

我开发的方糖猫平台的核心链路最后一步是视频渲染。用户提交生成任务后，系统需要把故事板、配音、素材合成为最终的视频文件。这个步骤 CPU/GPU 密集、耗时长、失败率高，单机渲染很快成为瓶颈。这篇文章会复盘我们从 HTTP 轮询架构迁移到 RocketMQ Push 架构的全过程。

一、渲染任务的特性

AI 视频平台的渲染任务有几个典型特征：

特性 影响
耗时长 一个 60 秒视频渲染可能需要 30 秒到 5 分钟
资源密集 CPU/GPU 占用高，单机并发有限
失败率高 素材缺失、字体缺失、内存不足、网络超时都可能失败
不均衡 不同分辨率、不同时长任务耗时差异大
需要重试 失败后最好换一台 Worker 重试，而不是一直卡在原机

这些特性决定了渲染不能放在主服务里同步执行，必须用异步任务队列 + 多 Worker 架构。

二、第一代架构：HTTP 轮询

2.1 设计思路

最早为了快速验证产品，我们用了最简单的方案：

sequenceDiagram
    participant A as API 服务
    participant W as Worker 服务
    A->>W: POST 任务
    A->>W: 查询状态
    W-->>A: 渲染完成
  • API 服务把渲染任务通过 HTTP POST 发给某个 Worker
  • Worker 开始渲染，API 服务定时轮询状态
  • 渲染完成后，Worker 回调 API 服务

2.2 代码示例

import requests
import time

class RenderService:
    def submit(self, task_id, render_params):
        worker_url = self.pick_worker()  # 简单轮询选一个
        requests.post(f"{worker_url}/render", json=render_params, timeout=5)

        # 轮询等待完成
        for _ in range(300):
            time.sleep(2)
            status = requests.get(f"{worker_url}/status/{task_id}").json()
            if status["done"]:
                return status["result"]

2.3 问题很快暴露

问题 说明
Worker 负载不均 轮询分配无法感知 Worker 当前负载
单点故障 某个 Worker 挂了，上面的任务就卡住
扩缩容麻烦 新增 Worker 需要改 API 服务配置
失败重试复杂 需要 API 服务记录每个任务发给了哪个 Worker
Windows 开发环境问题 开发环境只有一台机器，无法模拟多 Worker

这个方案支撑了产品早期验证，但当日均任务量超过几百条后，稳定性和扩展性都跟不上了。

三、第二代架构：RocketMQ Push

3.1 为什么选 RocketMQ

我们对比了几种消息队列：

方案 优点 缺点
HTTP 轮询 简单 负载不均、失败处理差
RabbitMQ 功能全面 高可用配置复杂
Kafka 吞吐高 消费端管理重，延迟不如 RocketMQ
RocketMQ 负载均衡好、顺序消息、延迟消息、监控完善 部署稍重

选择 RocketMQ 主要是因为：

  • 原生支持 Push Consumer，自动负载均衡
  • NameServer 架构适合弹性扩缩容
  • 消息重试、死信队列机制完善
  • 团队对阿里系中间件比较熟悉

3.2 新架构设计

flowchart TD
    A["API 服务"] -- "发送渲染任务" --> T["RocketMQ Topic: video-render"]
    T --> W1["Worker-1"]
    T --> W2["Worker-2"]
    T --> WN["Worker-N"]
    W1 --> D["渲染完成后更新数据库"]
    W2 --> D
    WN --> D
    D --> S["SSE 推送进度给用户"]

关键变化：

  • API 服务只负责发消息，不关心任务被哪个 Worker 消费
  • Worker 从 RocketMQ 拉取任务，自动负载均衡
  • 渲染完成后 Worker 更新数据库，并通过 SSE 通知前端
  • 新增 Worker 只需启动新实例，自动加入消费组

3.3 生产者实现

from rocketmq.client import Producer, Message

producer = Producer("render-producer-group")
producer.set_name_server_address("localhost:9876")
producer.start()

def submit_render_task(task_id: str, params: dict):
    body = json.dumps({
        "task_id": task_id,
        "params": params,
        "submitted_at": time.time()
    })
    msg = Message("video-render")
    msg.set_keys(task_id)
    msg.set_tags("normal")
    msg.set_body(body)
    producer.send_sync(msg)

3.4 消费者实现

from rocketmq.client import PushConsumer, ConsumeStatus

class RenderConsumer:
    def __init__(self):
        self.consumer = PushConsumer("render-consumer-group")
        self.consumer.set_name_server_address("localhost:9876")
        self.consumer.subscribe("video-render", "*", self.on_message)

    def on_message(self, msg):
        try:
            data = json.loads(msg.body)
            task_id = data["task_id"]
            params = data["params"]

            # 幂等检查
            if self.is_task_processed(task_id):
                return ConsumeStatus.CONSUME_SUCCESS

            # 更新任务状态为处理中
            self.update_task_status(task_id, "processing")

            # 执行渲染
            result = self.render(params)

            # 更新任务状态为完成
            self.update_task_status(task_id, "completed", result)

            return ConsumeStatus.CONSUME_SUCCESS
        except Exception as e:
            logger.error(f"Render failed: {e}")
            return ConsumeStatus.RECONSUME_LATER

    def start(self):
        self.consumer.start()

四、关键设计细节

4.1 幂等处理

消息队列可能重复投递，渲染任务又是耗时操作，必须做幂等：

def is_task_processed(task_id: str) -> bool:
    """基于数据库状态做幂等判断"""
    task = db.get_task(task_id)
    return task.status in ("completed", "processing")

如果任务已经是 processing 状态，说明可能被其他 Worker 在处理，需要进一步判断心跳时间，避免任务永远卡住。

4.2 心跳与任务认领

每个 Worker 在处理任务时会定时更新心跳：

def update_task_status(task_id, status, result=None):
    db.execute("""
        UPDATE render_task
        SET status = %s,
            result = %s,
            worker_id = %s,
            heartbeat_at = NOW()
        WHERE task_id = %s
    """, (status, result, WORKER_ID, task_id))

如果某个 Worker 挂了，心跳超过阈值的任务会被标记为失败，重新进入队列重试。

4.3 失败重试与死信队列

RocketMQ 支持消费失败自动重试，默认 16 次，间隔递增。如果还是失败，消息进入死信队列：

# 消费失败时返回重试
return ConsumeStatus.RECONSUME_LATER

对于死信消息，我们单独处理：

  • 分析失败原因
  • 通知运营或用户
  • 部分场景自动降级（比如降低分辨率重试）

4.4 并发控制

每个 Worker 同时能处理几个任务，取决于机器配置。我们通过线程池控制：

from concurrent.futures import ThreadPoolExecutor

# 根据 CPU/GPU 核心数设置
executor = ThreadPoolExecutor(max_workers=2)

def on_message(self, msg):
    future = executor.submit(self.handle_render, msg)
    # 注意：PushConsumer 是阻塞回调，这里需要等待结果
    result = future.result()
    return result

实际部署中，我们更推荐一个 Worker 进程只消费一个任务，通过启动多个 Worker 进程来提升并发。

五、Windows 开发环境兼容

5.1 问题

RocketMQ 在 Windows 上部署比较麻烦，开发环境有时候只有一台机器。为了不影响开发效率，我们设计了一个降级策略：

生产环境：RocketMQ Push Consumer
开发环境：HTTP 轮询模拟（单 Worker）

5.2 抽象层设计

class TaskConsumer(ABC):
    @abstractmethod
    def start(self):
        pass

class RocketMQConsumer(TaskConsumer):
    """生产环境使用"""
    pass

class HttpPollConsumer(TaskConsumer):
    """开发环境使用"""
    pass

# 根据配置切换
consumer = RocketMQConsumer() if env == "production" else HttpPollConsumer()

这样业务逻辑完全一致，只是消息消费方式不同。

六、监控与告警

6.1 关键指标

指标 说明
队列堆积量 判断是否需要扩容 Worker
任务平均处理时间 评估渲染性能和资源需求
任务成功率 健康度核心指标
Worker 在线数量 消费组健康状态
重试次数分布 发现系统性问题

6.2 告警规则

  • 队列堆积超过 100 条，持续 5 分钟 → 扩容提醒
  • 任务成功率低于 95% → 人工介入
  • Worker 数量异常减少 → 服务异常告警

七、演进效果

迁移到 RocketMQ 后，渲染系统的稳定性明显提升：

指标 HTTP 轮询 RocketMQ Push
平均等待时间 X 分钟 Y 分钟
任务成功率 X% Y%
Worker 利用率 不均 均衡
扩容成本 需要改配置 启动实例即可
失败重试 自建逻辑 原生支持

（待补充真实数据）

八、踩过的坑

8.1 消息体过大

早期的渲染参数里包含完整的素材 URL 列表，消息体很大。RocketMQ 默认单条消息限制 4MB，我们遇到过消息发送失败。

解决：消息体只传 task_id 和关键参数，Worker 消费后自己从数据库/redis 拉取完整参数。

8.2 消费组订阅不一致

不同 Worker 用同一个消费组但订阅了不同 Tag，导致消息分发混乱。

解决：严格规范消费组命名和订阅规则，每个消费组只负责一个 Topic。

8.3 任务卡住没有超时释放

某个 Worker 处理到一半进程挂了，任务状态一直是 processing。

解决：增加心跳检测，心跳超时的任务自动重置为 pending，重新入队。

九、总结

视频渲染是 AI 视频平台里最典型的「重任务」场景，必须用异步队列 + 分布式 Worker 来解决。几个关键经验：

  1. 不要过早优化，但也不能过度将就。HTTP 轮询能验证产品，但流量起来后必须升级。
  2. RocketMQ 的 Push Consumer 很适合这种场景，负载均衡和重试机制省去很多自建成本。
  3. 幂等和心跳是必须的，否则任务会丢或卡住。
  4. 监控要跟上，队列堆积和成功率是最核心的两个指标。

如果你也在做类似的异步任务处理系统，欢迎交流消息队列选型和 Worker 调度经验。

方糖猫
方糖猫（大可）

18 年后端架构，现在独立做 AI 产品。这里记录 AI 视频、AI Agent 与工程实践。

本文阅读量 次