React 项目中 消息队列 的设计取舍

最近在改公司的客服聊天模块，踩了一堆坑。起因很简单：测试提了个 bug，说快速发消息的时候偶尔会出现自己发的两条消息顺序颠倒，还有小概率某条消息直接消失。我第一反应是不可能，socket 本身是有序的啊。结果排查了两天发现，socket 确实有序，但消息从 onmessage 到 setState 再到渲染，中间被我们自己的代码搅得一塌糊涂。说白了，项目里根本没有消息队列这个概念，全是各组件各干各的，出问题纯属正常。

趁着重构，把消息队列的方案从头捋了一遍，这里把取舍过程记一下，没准儿能帮后来人省点时间。

先想清楚：你要的是队列，还是事件总线

这两个东西特别容易被混着用，我们项目就是重灾区。mitt 这种事件总线到处都是，一个事件 emit 出去，几个组件各订阅各的，看起来很松耦合，但问题是：没人管顺序，没人管丢了怎么办，也没人管消费者处理不过来怎么办。

队列和总线最大的区别是：总线是广播，发了就发了，他爱听不听；队列是有消费者承诺的，消息必须被处理，而且一般得按顺序。聊天消息、上传任务、埋点上报，这些属于队列；“用户点了一下按钮，好几个组件要响应”，这才是总线该干的事。想清楚这一点之后，我把聊天相关的事件全部从总线上摘了下来，这个决定后来证明是对的。

第一版：数组 + while 循环

一开始没想搞复杂，就写了个最朴素的版本：

class SimpleQueue {
  private tasks: (() => Promise<void>)[] = []
  private running = false

  push(task: () => Promise<void>) {
    this.tasks.push(task)
    this.drain()
  }

  private async drain() {
    if (this.running) return
    this.running = true
    while (this.tasks.length) {
      const task = this.tasks.shift()!
      await task()
    }
    this.running = false
  }
}

逻辑很简单：进队，没在跑就开始跑，跑完一个再拿下一个。本地跑测试一点问题没有，然后上线第一天就翻车了——有个任务抛了异常，await task() 直接炸出去，running 永远卡在 true，后面所有消息全部排队等死。这就是经典的“忘了 try/finally”，说实话挺丢人的，但这种坑不亲自踩一次真没感觉。

修好之后长这样：

private async drain() {
  if (this.running) return
  this.running = true
  try {
    while (this.tasks.length) {
      const task = this.tasks.shift()!
      try {
        await task()
      } catch (e) {
        console.error('task failed', e)
      }
    }
  } finally {
    this.running = false
  }
}

要不要去重和合并

第二个取舍是去重。消息进来的时候带个 key，同 key 的消息要么合并要么丢弃。比如“未读数更新”这种，一秒来 10 条你只关心最后一条，合并掉完全没问题；但聊天内容绝对不能去重，两条一样的消息也是两条。所以最后没做全局去重，而是在 push 的地方留了个参数：

push(task, { key?: string, merge?: 'replace' | 'skip' })

需要合并的场景自己声明，其他默认老老实实排队。这个设计不算优雅，但胜在显式，出问题好查。

要不要持久化

这是纠结最久的一个点。队列放内存里，用户一刷新全没了。埋点那边是真丢过数据的，所以埋点队列最后用了 IndexedDB 做持久化，上报成功再删；但聊天 UI 的队列就没必要了——刷新之后聊天记录本来就要从服务端重新拉，本地排队排得再好也是白排。说白了，要不要持久化取决于“这条消息丢了之后，服务端能不能帮你补回来”，能补就不用持久化，不能补才值得为此引入 IndexedDB 的序列化开销。

和 React 怎么相处

还有一个容易忽略的点：队列消费完之后怎么喂给 React。最开始我们是一条消息一次 setState，消息一密集，渲染直接卡成 PPT。后来改成 drain 一轮结束后统一 setState 一次，配合 React 18 的自动批处理，卡顿基本消失了。另外队列本身不要放进组件 state，它是个基础设施，放在模块级别的单例里，组件用 useSyncExternalStore 订阅长度和内容就够了。不然每条消息都触发一整套组件更新流程，纯属自己找罪受。

最后的取舍

最终方案大概就是：模块级单例队列，带 try/finally 的串行 drain，可选的 key 合并，聊天走内存、埋点走 IndexedDB。没上什么高大上的库，说实话这玩意儿自己写一百来行就够了，引第三方库反而要适配人家的生命周期和打包体积。当然，等消息量真上去之后这个方案大概率要改，比如要加优先级、要加背压策略，但那就是以后的事了，现在够用，先跑起来再说。