破局“音画不同步”：在线教育场景下TRTC信令与IM同步的极致优化实践

// Android端核心实现伪代码
public class SignalDispatcher {
    private SparseArray<Signal> pendingQueue = new SparseArray<>(); // 缓存乱序消息
    private int expectedSeq = 0; // 期望的下一条seq
    
    public void onReceiveIMessage(Message msg) {
        int seq = msg.getSeq();
        if (seq <= expectedSeq) {
            // 重复消息或过期消息，直接ACK丢弃
            return; 
        }
        
        if (seq == expectedSeq + 1) {
            // 顺序到达，立即分发
            dispatch(msg);
            expectedSeq++;
            // 尝试处理缓存队列中的下一条
            drainPendingQueue();
        } else {
            // 乱序到达（跳号），先缓存起来，等待前面的信令补全
            pendingQueue.put(seq, msg);
            // 启动超时定时器：若500ms内未补齐，则强制拉取云端历史消息补齐
            scheduleFetchHistory(seq);
        }
    }
}

// 前端（Web/小程序）伪代码
class MediaSignalCoordinator {
  constructor() {
    this.pendingActions = new Map(); // key: actionId, value: resolve
  }

  // 1. IM信令到达（如：批准连麦）
  onSignalArrive(signal) {
    if (signal.type === 'MIC_APPROVED') {
      // 不立即更新UI为“已上麦”，而是注册一个等待任务
      const actionId = signal.actionId;
      this.pendingActions.set(actionId, {
        status: 'WAITING_TRTC',
        timeout: setTimeout(() => this.forceApply(actionId), 3000) // 3秒超时兜底
      });
      
      // 调用TRTC SDK进入房间（如果还未进入）
      trtcClient.enterRoom({ roomId: signal.roomId, role: 'anchor' });
    }
  }

  // 2. TRTC 底层回调（真正的上麦成功）
  onTRTCEvent(event) {
    if (event.type === 'AUDIO_STATE_CHANGE' && event.state === 'ON') {
      const actionId = this.getPendingActionByUserId(event.userId);
      if (actionId && this.pendingActions.has(actionId)) {
        // 栅栏放行：此时才修改UI按钮状态为“已上麦”，并更新麦序
        this.updateUI('MIC_ON');
        clearTimeout(this.pendingActions.get(actionId).timeout);
        this.pendingActions.delete(actionId);
      }
    }
  }

  // 超时兜底（防止TRTC回调丢失）
  forceApply(actionId) {
    console.warn('TRTC回调超时，强制更新麦序');
    this.updateUI('MIC_ON');
    // 主动上报腾讯云CLS日志平台进行监控告警
    this.reportToCLS({ event: 'TIMEOUT_FALLBACK', actionId });
  }
}

// Golang 信令网关处理器（部署在TKE）
type Batcher struct {
    buckets map[string]*batch // roomId -> batch
    mu      sync.Mutex
}

func (b *Batcher) AddSignal(roomId string, signal []byte) {
    b.mu.Lock()
    defer b.mu.Unlock()
    if _, ok := b.buckets[roomId]; !ok {
        b.buckets[roomId] = &batch{signals: make([][]byte, 0, 10)}
    }
    b.buckets[roomId].signals = append(b.buckets[roomId].signals, signal)
}

// 定时器每20ms触发一次flush
func (b *Batcher) Flush(roomId string) {
    batch := b.buckets[roomId]
    if len(batch.signals) == 0 { return }
    
    // 压缩合并成一条IM消息（降低IM QPS消耗）
    merged := compressAndMerge(batch.signals) 
    
    // 调用腾讯云IM REST API发送群消息
    imClient.SendGroupMessage(roomId, merged)
    
    // 释放内存
    delete(b.buckets, roomId)
}

event:"TIMEOUT_FALLBACK" 
| select time, userId, roomId, actionId 
| order by time desc 
| limit 100