如何基于ExpressJS将事件处理系统暴露为REST服务API

数分钟级别的处理请求绝对不要用同步HTTP接口阻塞等待，不管是服务端连接开销、中间代理超时、网络波动都会导致极高的失败率，下面是经过工业界验证的成熟实现，完全适配你现有的Express封装层+RabbitMQ架构，改造成本极低。

方案1：异步任务模式（兼容性首选，业界标准实现） #

这是各大云厂商、SaaS服务做长耗时API的通用方案，没有任何协议依赖，所有类型客户端都能对接：

• 核心流程：
  • 客户端提交处理请求时，接口立刻返回202 Accepted状态码，响应体直接返回taskId——直接复用你现有系统的Message ID即可，不需要额外生成唯一标识
  • 提交接口支持可选传callbackUrl参数：如果客户端传了这个地址，等RabbitMQ侧返回处理结果后，Express封装层主动发POST请求把结果推到该地址；回调加3次指数退避重试，重试失败就标记回调失败，留存结果供客户端主动拉取
  • 配套两个轻量接口：GET /tasks/:taskId返回任务当前状态（排队中/处理中/处理成功/处理失败），成功/失败时直接携带结果；如果结果体很大，可以额外拆一个GET /tasks/:taskId/result接口单独返回结果，减少状态查询的带宽开销
  • 状态查询接口必须返回Retry-After响应头，比如排队中返回Retry-After: 10、处理中返回Retry-After: 30，告诉客户端多久之后再来查，从接口层面避免无意义的高频轮询打崩服务
• 适配现有架构的改造点：
  • Express收到请求后，透传Message ID投递消息到RabbitMQ即可，不要同步等待结果，把taskId和初始状态存入轻量存储（量级小直接用内存Map，量级大上Redis做缓存即可）
  • 单独起一个常驻的消费者进程，监听你现有的结果输出队列，收到结果后更新对应taskId的状态、缓存结果，触发回调推送逻辑
  • 结果缓存根据业务需求留存1-24小时即可，避免客户端首次拉取失败后无法获取结果
• 优缺点：兼容性拉满，不管客户端是后端服务、前端页面、老旧系统都能对接，稳定性最高；缺点是未传回调地址的客户端需要主动轮询，但通过Retry-After头可以把轮询压力降到极低。

方案2：HTTP长连接推送（零轮询体验，适合可控客户端） #

如果你的客户端是可控的（比如内部系统、官方SDK对接的客户），可以用SSE（Server-Sent Events）替代轮询，改造成本极低，完全不需要客户端主动查询进度：

• 核心流程：
  • 客户端提交任务拿到taskId后，直接请求GET /tasks/:taskId/stream的SSE接口，和服务端建立HTTP长连接
  • 等RabbitMQ返回处理结果后，Express层直接通过这个长连接把结果推给客户端，推送完成后主动关闭连接
  • 服务端定期发心跳包（比如每15秒发一个空注释行）避免中间代理掐断空闲连接，同时设置10分钟左右的连接超时，超时后主动断开并提示客户端重新连接或走状态查询接口拿结果
• 为什么不用WebSocket？SSE是纯HTTP协议，比WebSocket轻量太多，不需要额外做协议升级、连接保活的复杂逻辑，浏览器原生支持，后端服务调用也有成熟的客户端实现；只有当你的服务本身已经在维护WebSocket连接池的时候，才考虑用WebSocket推结果。
• 极简实现参考：

// 维护taskId到SSE连接的映射
const sseConnections = new Map()

app.get('/tasks/:taskId/stream', async (req, res) => {
  const { taskId } = req.params
  // 先查缓存，任务已经处理完直接返回结果
  const cachedTask = await taskCache.get(taskId)
  if (cachedTask?.status === 'completed' || cachedTask?.status === 'failed') {
    res.writeHead(200, {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache'
    })
    res.write(`data: ${JSON.stringify(cachedTask)}\n\n`)
    return res.end()
  }
  // 建立SSE连接
  res.writeHead(200, {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
    'Connection': 'keep-alive'
  })
  sseConnections.set(taskId, res)
  // 心跳保活
  const heartbeat = setInterval(() => res.write(': heartbeat\n\n'), 15000)
  // 超时处理
  req.setTimeout(10 * 60 * 1000, () => {
    clearInterval(heartbeat)
    sseConnections.delete(taskId)
    res.write(`event: timeout\ndata: {"msg":"processing timeout, please query status later"}\n\n`)
    res.end()
  })
  // 客户端断连清理
  req.on('close', () => {
    clearInterval(heartbeat)
    sseConnections.delete(taskId)
  })
})

// 结果消费者拿到结果后，推给对应SSE连接
function onTaskResult(taskId, result) {
  taskCache.set(taskId, result, 24 * 60 * 60) // 缓存24小时
  const conn = sseConnections.get(taskId)
  if (conn) {
    conn.write(`data: ${JSON.stringify(result)}\n\n`)
    conn.end()
    sseConnections.delete(taskId)
  }
  // 触发回调逻辑
  triggerCallbackIfNeeded(taskId, result)
}

• 优缺点：完全不需要客户端轮询，实时性好，连接开销远低于反复轮询；缺点是部分严格的企业网络代理会拦截长时间空闲的HTTP连接，对不可控的第三方客户端兼容性不如方案1。

明确不推荐的做法 #

• 不要用普通POST接口同步阻塞等待结果：就算把Express、Node的超时设得再长，中间的Nginx、CDN、客户端超时配置、网络波动都会导致数分钟级的请求失败率居高不下
• 不要放任客户端无限制轮询：必须通过Retry-After头引导客户端按合理频率查询，否则排队任务多的时候轮询请求会占满服务带宽
• 不要消费完RabbitMQ的结果就直接丢弃：一定要做结果缓存，避免客户端因为网络问题没拿到结果就再也找不到处理数据

选型建议：如果要对外给未知的第三方客户开放，直接选方案1，回调+状态查询的组合是当前长耗时OpenAPI的事实标准；如果客户端都是内部系统或者可控的合作方，直接上SSE，开发量加起来不到100行代码，体验比轮询好很多。

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