WebRTC+MQTT异步回调异常问题排查与解决方案问询

问题场景 #

我们采用MQTT作为WebRTC的信令协议，被叫流程的伪代码如下：

mqttClient.on('message', messageHandler)

async messageHandler(peerConnection, topic, message) {
  switch (message.type) {
    case 'offer':
      const stream = await getUserMedia({ video: true })
      stream.getTracks().forEach(track => peerConnection.addTrack(track, stream))
      await peerConnection.setRemoteDescription(message.offer)
      const answer = await peerConnection.createAnswer()
      peerConnection.setLocalDescription(answer)
      mqttClient.publish(destinationTopic, answer)
      break
    case 'candidate':
      peerConnection.addIceCandidate(message.candidate);
  }
}

标准的WebRTC被叫流程应该是：

• 接收offer
• 添加媒体流
• 设置远端描述
• 创建answer
• 设置本地描述
• 发送answer
• 接收candidate

但实际运行时触发WebRTC API报错：setRemoteDescription needs to called before addIceCandidate

问题根源 #

问题出在await getUserMedia({ video: true })这一步：

• 收到offer后调用getUserMedia，这是耗时较长的异步操作，会暂停当前messageHandler的执行
• MQTT的消息回调不会等待异步回调执行完成，此时如果candidate消息到达，会直接再次调用messageHandler处理candidate
• 此时处理offer的流程还卡在获取媒体流阶段，setRemoteDescription还没执行，直接调用addIceCandidate就会触发报错

疑问解答 #

1. 为何MQTT.js及JS标准API不等待前一个异步回调完成？ #

事件驱动类API（比如MQTT.js的message事件、Node.js EventEmitter）的设计核心是快速响应事件，如果强制等待前一个异步回调完成，会导致事件队列积压，拖慢系统响应速度，违背了JS异步非阻塞的设计原则。

forEach/map/reduce这类数组方法本身是同步设计，它们的定位是遍历处理数组元素，并不负责管理异步任务的执行顺序，因此也不会等待异步回调完成。

2. 为何addIceCandidate必须在setRemoteDescription之后调用？ #

ICE候选是基于SDP（会话描述协议）生成的，setRemoteDescription会把远端的会话信息（包括媒体能力、网络参数等）告知WebRTC。只有拿到这些上下文信息，WebRTC才能判断ICE候选是否匹配当前会话、如何建立网络连接。如果先调用addIceCandidate，WebRTC没有远端会话的上下文，无法处理这些候选，因此会抛出错误。

解决方案 #

1. 临时修复方案 #

调整流程顺序，先执行setRemoteDescription，再获取媒体流，同时添加候选缓存逻辑，确保candidate处理时上下文已就绪：

async messageHandler(peerConnection, topic, message) {
  switch (message.type) {
    case 'offer':
      // 优先设置远端描述，保证candidate处理的上下文
      await peerConnection.setRemoteDescription(message.offer)
      // 处理缓存的候选（如果有的话）
      if (peerConnection.pendingCandidates) {
        for (const candidate of peerConnection.pendingCandidates) {
          await peerConnection.addIceCandidate(candidate)
        }
        delete peerConnection.pendingCandidates
      }
      // 再获取媒体流并添加
      const stream = await getUserMedia({ video: true })
      stream.getTracks().forEach(track => peerConnection.addTrack(track, stream))
      // 后续流程不变
      const answer = await peerConnection.createAnswer()
      peerConnection.setLocalDescription(answer)
      mqttClient.publish(destinationTopic, answer)
      break
    case 'candidate':
      // 如果远端描述已设置，直接处理候选；否则缓存起来
      if (peerConnection.remoteDescription) {
        peerConnection.addIceCandidate(message.candidate);
      } else {
        peerConnection.pendingCandidates = peerConnection.pendingCandidates || []
        peerConnection.pendingCandidates.push(message.candidate)
      }
  }
}

这种调整能确保即使candidate在获取媒体流阶段到达，要么直接处理，要么被缓存后统一处理，不会触发报错。

2. 长期最优解决方案 #

引入异步任务队列+状态管理，为每个PeerConnection维护串行的任务队列，保证信令消息按顺序处理：

// 用WeakMap为每个PeerConnection绑定独立的任务队列
const peerTaskQueues = new WeakMap()

async function enqueuePeerTask(peerConnection, task) {
  let queue = peerTaskQueues.get(peerConnection)
  if (!queue) {
    queue = []
    peerTaskQueues.set(peerConnection, queue)
  }
  queue.push(task)
  // 如果队列之前是空的，立即启动串行执行
  if (queue.length === 1) {
    while (queue.length > 0) {
      await queue[0]()
      queue.shift()
    }
  }
}

mqttClient.on('message', (topic, message) => {
  // 假设通过topic或message能找到对应的PeerConnection
  const peerConnection = getPeerConnectionFromTopic(topic)
  // 将所有信令处理逻辑加入队列，保证串行执行
  enqueuePeerTask(peerConnection, async () => {
    switch (message.type) {
      case 'offer':
        const stream = await getUserMedia({ video: true })
        stream.getTracks().forEach(track => peerConnection.addTrack(track, stream))
        await peerConnection.setRemoteDescription(message.offer)
        const answer = await peerConnection.createAnswer()
        peerConnection.setLocalDescription(answer)
        mqttClient.publish(destinationTopic, answer)
        break
      case 'candidate':
        peerConnection.addIceCandidate(message.candidate);
        break
    }
  })
})

这种方案从根本上解决了异步回调的顺序问题，适合复杂的WebRTC信令场景，能避免因消息乱序导致的各类API报错。

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