31 KiB

Raw Blame History

任务队列后端 - message-dispatcher 优先分发技术方案

1. 项目概述

1.1 背景

当前任务队列后端系统通过 runninghub 等外部平台处理 comfyui 任务。为降低成本、提高响应速度，需要实现内部算力优先使用机制，将 runninghub 类型任务优先路由至内部 message-dispatcher 系统。

1.2 目标

实现 comfyui 类型任务优先分发至内部 message-dispatcher
确保当内部算力不可用时自动降级至外部平台
保持接口兼容性，不影响现有功能
提供完善的健康检查和监控机制

2. 系统架构分析

2.1 任务队列后端架构

任务队列后端 (任务队列后端/)
├── index.js                    # HTTP 回调服务 (端口 8066)
├── outside/
│   ├── generat.js             # 任务分发逻辑
│   └── outPlatforms/
│       ├── outside.js         # 平台管理
│       └── runninghub.js      # RunningHub 平台实现
├── worker_threads/
│   └── wait/
│       ├── waiting.js         # 任务队列轮询
│       └── generatTask.js     # 任务处理 Worker
└── redis/                      # Redis 数据存储

2.2 message-dispatcher 架构

message-dispatcher/
├── src/
│   ├── index.js                # 主服务 (端口 4000)
│   ├── api/
│   │   └── index.js           # REST API 路由
│   ├── bridge-manager/        # 桥接器管理
│   └── websocket-server/      # WebSocket 服务

2.3 关键接口分析

任务队列后端 - 任务流程：

任务进入 Redis 等待队列
waiting.js 轮询获取任务
调用 generatTask.js → externalPostRequest()
通过 outside[runninghub] 发送至外部平台

message-dispatcher - 任务接口：

POST /api/task - 提交任务
GET /api/health - 健康检查
GET /api/overview - 系统概览

3. 方案设计

3.1 总体架构

任务队列后端 ──┐
               │ 优先
               ▼
      message-dispatcher (内部算力)
               │
               └─ 失败/不可用时
               │
               ▼
         runninghub (外部平台)

3.2 核心模块设计

3.2.1 新增模块：内部算力客户端 (InternalDispatcher)

文件位置： 任务队列后端/outside/outPlatforms/messageDispatcher.js

功能：

封装 message-dispatcher API 调用
提供与 runninghub 一致的接口方法
仅修改请求地址和请求头，其他配置与 runninghub 完全一致

接口方法：

getGenerateUrl()           // 获取 API 地址
getGenerateHeader(apikey)  // 获取请求头（包含 JWT Token）
getGenerateBody(task)      // 构造请求体（与 runninghub 完全一致）
getSuccessTasks(response)  // 处理成功响应
getTaskResult(response)    // 处理结果回调
getQueryUrl()              // 获取回调地址

3.2.2 健康检查模块

文件位置： 任务队列后端/utils/healthCheck.js

功能：

定期检查 message-dispatcher 健康状态
维护可用状态缓存
提供可用性查询接口
监控内部算力可用数量

检查项：

HTTP 接口连通性 (/api/health)
可用桥接器数量
在线实例数量（内部算力可用数）

3.2.3 路由决策模块

修改位置： 任务队列后端/outside/generat.js 的 externalPostRequest()

决策逻辑：

获取内部算力可用数 (internalCapacity)
获取 runninghub 可容纳上限 (externalCapacity)
总可分发任务上限 = internalCapacity + externalCapacity

if (待分发任务数 ≤ internalCapacity) {
  全部任务使用内部算力
} else {
  前 internalCapacity 个任务使用内部算力
  超出部分使用 runninghub（不超过 externalCapacity）
}

3.3 接口对齐方案

3.3.1 配置对齐要求

重要原则：

message-dispatcher 配置与 runninghub 保持完全一致
仅修改以下两个参数：
1. 请求地址：从 runninghub URL 改为 message-dispatcher URL
2. 请求头：将 apiKey 字段值替换为 JWT Token
其他所有配置项、参数、字段名称均不做任何改动

3.3.2 请求体标准化

请求体必须与 runninghub 完全一致，包括：

字段名称	类型	说明	是否修改
`workflow_id`	String	工作流 ID	否
`node_info_list`	Array	节点信息列表	否
`apiKey`	String	API Key	是（值改为 JWT Token）
`webhookUrl`	String	回调地址	否
其他所有字段	-	保持原样	否

示例对比：

// runninghub 请求体（原样）
{
  "workflow_id": "123",
  "node_info_list": [...],
  "apiKey": "runninghub-api-key-xxx",
  "webhookUrl": "http://callback-url"
}

// message-dispatcher 请求体（仅修改 apiKey 值）
{
  "workflow_id": "123",
  "node_info_list": [...],
  "apiKey": "eyJhbGciOiJIUzI1NiIs...",  // JWT Token
  "webhookUrl": "http://callback-url"
}

3.3.3 响应格式转换

message-dispatcher → 任务队列后端格式映射：

// message-dispatcher 响应
{
  success: true,
  data: {
    requestId: "uuid",
    taskId: "..."
  }
}

// 转换为 runninghub 兼容格式
{
  msg: "success",
  code: 0,
  data: {
    taskId: "requestId"
  }
}

3.4 降级机制

3.4.1 降级触发条件

message-dispatcher 健康检查失败
任务提交请求超时 (30秒)
任务提交返回错误
无可用桥接器/实例（内部算力可用数为0）

3.4.2 降级流程

记录降级日志
标记本次任务使用外部平台
更新健康检查状态
继续使用 runninghub 完成任务

3.5 JWT 认证机制

3.5.1 JWT 生成与发送

JWT 获取流程：

初始登录：
- 调用 message-dispatcher 登录接口
- 接口：POST /api/auth/login
- 请求体：{ username: "admin", password: "2233..2233" }
- 获取 access_token
JWT 存储：
- 存储在内存变量中
- 同时存储过期时间（expires_at）
请求头设置：
- 将 JWT Token 作为 apiKey 字段的值
- 请求头格式与 runninghub 保持一致

3.5.2 JWT 定期更新机制

更新策略：

更新间隔：20小时（默认 JWT 有效期 24小时，提前4小时更新）
更新流程：
1. 检查当前 Token 是否在 4 小时内过期
2. 如果是，重新调用登录接口获取新 Token
3. 更新内存中的 Token 和过期时间
4. 记录 Token 更新日志

实现代码：

class JWTManager {
  constructor() {
    this.token = null;
    this.expiresAt = null;
    this.refreshInterval = 20 * 60 * 60 * 1000; // 20小时
  }

  async init() {
    await this.refreshToken();
    setInterval(() => this.refreshToken(), this.refreshInterval);
  }

  async refreshToken() {
    try {
      const response = await fetch(`${MESSAGE_DISPATCHER_URL}/api/auth/login`, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({
          username: process.env.MD_USERNAME || 'admin',
          password: process.env.MD_PASSWORD || '2233..2233'
        })
      });
      
      const data = await response.json();
      if (data.success && data.data?.access_token) {
        this.token = data.data.access_token;
        this.expiresAt = Date.now() + 24 * 60 * 60 * 1000; // 假设24小时有效期
        console.log('[JWTManager] Token 已更新');
      }
    } catch (error) {
      console.error('[JWTManager] Token 更新失败:', error);
    }
  }

  getToken() {
    return this.token;
  }

  isTokenValid() {
    return this.token && Date.now() < this.expiresAt;
  }
}

3.6 任务分流逻辑

3.6.1 核心计算公式

内部算力可用数 = message-dispatcher 在线实例数
runninghub 可容纳上限 = 配置的外部平台并发数
总可分发任务上限 = 内部算力可用数 + runninghub 可容纳上限

3.6.2 具体分发策略

场景 1：待分发任务数 ≤ 内部算力可用数

示例：
内部算力可用数 = 30
待分发任务数 = 25

结果：
全部 25 个任务使用内部算力处理

场景 2：待分发任务数 > 内部算力可用数

示例：
内部算力可用数 = 30
runninghub 可容纳上限 = 10
待分发任务数 = 45

结果：
前 30 个任务 → 内部算力
后 10 个任务 → runninghub（达到上限）
剩余 5 个任务 → 等待队列（等待资源释放）

3.6.3 分发决策实现

修改位置： worker_threads/wait/waiting.js 和 worker_threads/wait/generatTask.js

实现逻辑：

async function distributeTasks(tasks) {
  const internalCapacity = await getInternalCapacity();
  const externalCapacity = await getExternalCapacity();
  
  const internalTasks = tasks.slice(0, internalCapacity);
  const externalTasks = tasks.slice(internalCapacity, internalCapacity + externalCapacity);
  
  // 分发到内部算力
  for (const task of internalTasks) {
    await sendToMessageDispatcher(task);
  }
  
  // 分发到外部平台
  for (const task of externalTasks) {
    await sendToRunningHub(task);
  }
  
  // 剩余任务返回队列
  const remainingTasks = tasks.slice(internalCapacity + externalCapacity);
  return remainingTasks;
}

3.7 配置设计

新增环境变量 (.env)：

# Message Dispatcher 配置
MESSAGE_DISPATCHER_URL=http://localhost:4000
MESSAGE_DISPATCHER_ENABLED=true
MESSAGE_DISPATCHER_TIMEOUT=30000
MESSAGE_DISPATCHER_HEALTH_CHECK_INTERVAL=10000

# Message Dispatcher 认证
MD_USERNAME=admin
MD_PASSWORD=2233..2233

# 容量配置
INTERNAL_CAPACITY_MAX=30
EXTERNAL_CAPACITY_MAX=10

新增配置文件 (config/messageDispatcher.json)：

{
  "enabled": true,
  "priority": true,
  "healthCheck": {
    "interval": 10000,
    "timeout": 5000,
    "failureThreshold": 3
  },
  "task": {
    "timeout": 30000,
    "retryCount": 1
  },
  "capacity": {
    "internal": 30,
    "external": 10
  },
  "jwt": {
    "refreshInterval": 72000000
  }
}

4. 实现步骤

步骤 1: 创建 message-dispatcher 平台适配器

创建 outside/outPlatforms/messageDispatcher.js
实现与 runninghub.js 相同的接口
请求体与 runninghub 完全一致，仅修改 apiKey 值为 JWT Token

步骤 2: 实现 JWT 认证模块

创建 utils/jwtManager.js
实现 Token 获取、存储、定期更新逻辑
集成到启动流程

步骤 3: 实现健康检查模块

创建 utils/healthCheck.js
实现定期健康检查逻辑
监控内部算力可用数量

步骤 4: 修改任务分发逻辑

修改 outside/generat.js - 单任务分发决策
修改 worker_threads/wait/waiting.js - 批量任务分流
根据内部/外部容量进行任务分配

步骤 5: 更新平台管理

修改 outside/outPlatforms/outside.js
注册 messageDispatcher 平台

步骤 6: 添加日志和监控

新增分发决策日志
新增容量使用统计日志
新增降级统计日志

步骤 7: 测试验证

单元测试
集成测试
降级场景测试
容量边界测试

5. 验收标准

5.1 功能验收

任务分流逻辑正确：≤内部容量全部走内部，超出部分走外部
message-dispatcher 不可用时自动降级至 runninghub
请求体与 runninghub 完全一致（仅 apiKey 值不同）
JWT Token 自动更新机制正常工作
任务成功执行并返回正确结果
回调接口正常工作
现有功能不受影响

5.2 接口兼容性验收

messageDispatcher.js 接口方法签名与 runninghub.js 完全一致
请求体字段名称、类型、格式与 runninghub 完全一致
响应格式与 runninghub 兼容
不修改 runninghub.js 的任何代码

5.3 性能验收

健康检查延迟 < 1s
任务分发决策时间 < 100ms
降级切换时间 < 1s
系统整体吞吐量不降低

5.4 可靠性验收

连续 24 小时稳定运行
降级机制在故障场景下正确触发
JWT Token 自动更新不中断服务
日志完整可追溯
错误处理完善

6. 风险与应对

风险	影响	概率	应对措施
message-dispatcher 接口变更	高	中	版本兼容检查，完善监控
JWT Token 过期	高	低	提前更新，多重校验
任务格式不匹配	高	低	充分测试，格式验证
降级失败	高	低	多级降级，超时保护
性能下降	中	低	性能测试，优化热点路径

7. WebSocket 双向通信机制

7.1 架构设计

建立任务队列后端与 message-dispatcher 之间的持久化 WebSocket 连接，实现实时双向通信。

任务队列后端 (WebSocket 客户端)
        │
        │ 建立连接
        ▼
message-dispatcher (WebSocket 服务端)

7.2 连接管理

7.2.1 连接建立流程

任务队列后端启动时：
- 初始化 WebSocket 客户端
- 连接至 message-dispatcher 的 WebSocket 端点
- 发送 CONNECT 消息进行身份认证
- 接收 CONNECT_ACK 确认连接建立
消息格式定义：

// 连接消息
{
  type: 'CONNECT',
  data: {
    clientId: 'task-queue-backend',
    timestamp: '2024-01-01T00:00:00.000Z'
  }
}

// 连接确认
{
  type: 'CONNECT_ACK',
  data: {
    serverId: 'message-dispatcher',
    timestamp: '2024-01-01T00:00:00.000Z'
  }
}

7.2.2 心跳保活机制

心跳间隔：30秒
超时时间：10秒
重连策略：指数退避（1s, 2s, 4s, 8s, 16s, 最大32s）

// 心跳消息
{
  type: 'HEARTBEAT',
  data: {
    timestamp: '2024-01-01T00:00:00.000Z'
  }
}

// 心跳响应
{
  type: 'HEARTBEAT_ACK',
  data: {
    timestamp: '2024-01-01T00:00:00.000Z'
  }
}

7.3 算力状态上报

7.3.1 上报内容

message-dispatcher 定期向任务队列后端上报以下信息：

{
  type: 'CAPACITY_UPDATE',
  data: {
    timestamp: '2024-01-01T00:00:00.000Z',
    bridges: [
      {
        bridgeId: 'bridge-1',
        connectedAt: '2024-01-01T00:00:00.000Z',
        instances: [
          {
            instanceId: 'instance-1',
            status: 'online', // online | busy | offline
            currentTask: null | { taskId: 'xxx', startTime: '...' }
          }
        ]
      }
    ],
    summary: {
      totalBridges: 2,
      totalInstances: 8,
      onlineInstances: 6,
      busyInstances: 2,
      availableCapacity: 4  // 可同时处理的任务数
    }
  }
}

7.3.2 上报触发条件

定期上报：每 10 秒上报一次
事件触发：
- 桥接器连接/断开
- 实例状态变化（online → busy, busy → online, offline）
- 任务开始/完成

7.4 任务数量下发

任务队列后端向 message-dispatcher 下发任务分配信息：

{
  type: 'TASK_ALLOCATION',
  data: {
    timestamp: '2024-01-01T00:00:00.000Z',
    allocatedTasks: [
      {
        taskId: 'task-1',
        targetInstance: 'instance-1'
      }
    ]
  }
}

7.5 实例状态变化监测

7.5.1 变化事件类型

// 实例上线
{
  type: 'INSTANCE_ONLINE',
  data: {
    bridgeId: 'bridge-1',
    instanceId: 'instance-1',
    timestamp: '2024-01-01T00:00:00.000Z'
  }
}

// 实例下线
{
  type: 'INSTANCE_OFFLINE',
  data: {
    bridgeId: 'bridge-1',
    instanceId: 'instance-1',
    timestamp: '2024-01-01T00:00:00.000Z',
    reason: 'disconnected'
  }
}

// 实例繁忙
{
  type: 'INSTANCE_BUSY',
  data: {
    bridgeId: 'bridge-1',
    instanceId: 'instance-1',
    taskId: 'task-1',
    timestamp: '2024-01-01T00:00:00.000Z'
  }
}

// 实例空闲
{
  type: 'INSTANCE_IDLE',
  data: {
    bridgeId: 'bridge-1',
    instanceId: 'instance-1',
    completedTaskId: 'task-1',
    timestamp: '2024-01-01T00:00:00.000Z'
  }
}

8. Redis 状态管理机制

8.1 Redis 数据结构设计

8.1.1 连接状态存储

Key 前缀： md:state:

Key	类型	说明	TTL
`md:state:connection`	Hash	连接状态信息	永久
`md:state:capacity`	Hash	算力容量信息	60s
`md:state:instances`	Hash	实例详细状态	60s
`md:state:last_heartbeat`	String	最后心跳时间	60s

connection Hash 字段：

{
  status: 'connected',      // connected | disconnected
  connectedAt: '2024-01-01T00:00:00.000Z',
  disconnectedAt: null,
  disconnectReason: null,
  reconnectAttempts: 0
}

capacity Hash 字段：

{
  totalBridges: 2,
  totalInstances: 8,
  onlineInstances: 6,
  busyInstances: 2,
  availableCapacity: 4,
  updatedAt: '2024-01-01T00:00:00.000Z'
}

instances Hash 结构： Key: md:state:instances:{instanceId}

{
  instanceId: 'instance-1',
  bridgeId: 'bridge-1',
  status: 'online',
  currentTaskId: null,
  currentTaskStartTime: null,
  lastUpdated: '2024-01-01T00:00:00.000Z'
}

8.2 健康检查机制

8.2.1 检查维度

WebSocket 连接检查
- 检查连接状态是否为 connected
- 检查最后心跳时间是否在 60 秒内
算力可用性检查
- 检查在线实例数量 > 0
- 检查可用容量 > 0
综合健康评分

健康状态 = 连接状态 × 0.4 + 算力状态 × 0.6

8.2.2 健康检查流程

async function checkHealth() {
  const state = await getConnectionState();
  const capacity = await getCapacity();
  
  const isConnected = state.status === 'connected';
  const hasHeartbeat = Date.now() - state.lastHeartbeat < 60000;
  const hasCapacity = capacity.availableCapacity > 0;
  
  const isHealthy = isConnected && hasHeartbeat && hasCapacity;
  
  if (!isHealthy) {
    await triggerDegradation();
  }
  
  return isHealthy;
}

8.3 自动降级触发

8.3.1 降级触发条件

条件	说明	优先级
WebSocket 断开	连接完全断开	1
心跳超时 (>60s)	连接可能异常	2
无可用实例	在线实例数为0	3
可用容量为0	所有实例都忙	4

8.3.2 降级流程

检测到异常
更新 Redis 状态：设置 status: 'degraded'
记录降级日志：记录降级时间、原因、当前状态
通知路由模块：触发路由策略切换
持续监控：定期检查是否恢复

8.3.3 自动恢复流程

健康检查通过
等待稳定期：连续 3 次健康检查通过（30秒）
更新 Redis 状态：设置 status: 'healthy'
记录恢复日志
通知路由模块：恢复优先使用内部算力

9. Message-Dispatcher 任务缓存与调度机制

9.1 缓存队列设计

9.1.1 缓存队列数据结构

使用 Redis List 实现：

Key	类型	说明
`md:cache:pending`	List	待执行任务缓存
`md:cache:processing`	Hash	执行中任务
`md:cache:completed`	List	已完成任务（最近1000条）

pending List 元素格式：

{
  taskId: 'xxx',
  payload: {...},
  submittedAt: '2024-01-01T00:00:00.000Z',
  retryCount: 0
}

processing Hash 字段：

{
  taskId: {
    instanceId: 'instance-1',
    bridgeId: 'bridge-1',
    startTime: '2024-01-01T00:00:00.000Z',
    payload: {...}
  }
}

9.2 任务调度器

9.2.1 调度器架构

调度器主循环
    │
    ├─ 获取可用实例
    │
    ├─ 从缓存队列取出任务
    │
    ├─ 分配任务至实例
    │
    └─ 移动至 processing

9.2.2 调度算法

async function scheduleTasks() {
  const availableInstances = await getAvailableInstances();
  const pendingTasks = await getPendingTasks(availableInstances.length);
  
  for (let i = 0; i < pendingTasks.length; i++) {
    const task = pendingTasks[i];
    const instance = availableInstances[i];
    
    await assignTask(task, instance);
  }
}

9.2.3 实例选择策略

最少任务优先：选择当前任务数最少的实例
同桥接优先：优先选择同一桥接器下的实例
随机打散：避免总是选择同一实例

function selectOptimalInstance(availableInstances, task) {
  // 按当前任务数排序
  const sorted = [...availableInstances].sort((a, b) => 
    a.currentTasks - b.currentTasks
  );
  
  // 如果有任务指定偏好实例
  if (task.preferredInstanceId) {
    const preferred = sorted.find(i => i.id === task.preferredInstanceId);
    if (preferred) return preferred;
  }
  
  return sorted[0];
}

9.3 任务完成监听

9.3.1 完成事件处理

async function handleTaskComplete(taskId, result) {
  // 1. 从 processing 移除
  const processingInfo = await removeFromProcessing(taskId);
  
  // 2. 标记实例为空闲
  await markInstanceIdle(processingInfo.instanceId);
  
  // 3. 添加到 completed
  await addToCompleted(taskId, result);
  
  // 4. 触发新一轮调度
  await scheduleTasks();
  
  // 5. 调用回调（如果有）
  if (processingInfo.webhookUrl) {
    await invokeWebhook(processingInfo.webhookUrl, result);
  }
}

9.3.2 失败重试机制

async function handleTaskFailure(taskId, error) {
  const task = await getTask(taskId);
  const maxRetries = 3;
  
  if (task.retryCount < maxRetries) {
    task.retryCount++;
    task.lastError = error.message;
    await returnToPending(task);
  } else {
    await markAsFailed(taskId, error);
  }
}

9.4 缓存队列管理

9.4.1 队列扩容/缩容

当实例数量变化时：

async function adjustCacheQueue(instanceDelta) {
  const currentPending = await getPendingCount();
  const currentCapacity = await getAvailableCapacity();
  
  if (instanceDelta > 0) {
    // 实例增加，立即调度
    await scheduleTasks();
  } else if (instanceDelta < 0) {
    // 实例减少，检查是否需要回退任务
    const excess = currentPending - currentCapacity;
    if (excess > 0) {
      // 通知任务队列后端，这些任务需要降级
      await notifyTaskQueueForDegradation(excess);
    }
  }
}

9.4.2 超时清理

async function cleanStaleTasks() {
  const now = Date.now();
  const staleThreshold = 30 * 60 * 1000; // 30分钟
  
  const pendingTasks = await getAllPendingTasks();
  
  for (const task of pendingTasks) {
    if (now - task.submittedAt > staleThreshold) {
      await removeFromPending(task.taskId);
      await markAsExpired(task.taskId);
    }
  }
}

10. 任务队列逻辑健壮性测试方案

10.1 测试场景设计

10.1.1 场景1：内部算力减少

测试目的： 验证当内部算力减少时，系统能否正确处理已下发任务

测试步骤：

初始状态：4个在线实例，下发4个任务
操作：突然断开2个实例
预期结果：
- 未开始的任务重新回到缓存队列
- 正在运行的任务继续执行（如果实例还在）
- 系统记录日志，不崩溃
- 自动调度可用实例处理缓存任务

测试用例代码：

async function testCapacityReduction() {
  // 1. 准备环境
  await setupTestEnvironment({ onlineInstances: 4 });
  
  // 2. 下发任务
  const taskIds = await submitTasks(4);
  
  // 3. 确认任务开始执行
  await waitForTasksRunning(taskIds);
  
  // 4. 断开2个实例
  await disconnectInstances(2);
  
  // 5. 验证
  const remainingTasks = await getRemainingTasks();
  assert(remainingTasks.length <= 2, '不应有超过2个任务继续执行');
  
  const cacheQueue = await getCacheQueue();
  assert(cacheQueue.length >= 2, '应有至少2个任务回到缓存');
  
  const systemHealth = await checkSystemHealth();
  assert(systemHealth.status === 'degraded', '系统应处于降级状态');
  assert(systemHealth.crashed === false, '系统不应崩溃');
}

10.1.2 场景2：任务数 > 可用实例数

测试目的： 验证当上一次下发任务数大于当前可用任务数时的系统行为