把 FCM 改成 Topic：代价是要把订阅关系维护对

我做过一个 App 后端的推送系统，基于 Node.js、TypeScript 和 Prisma。运营人员在后台发推送，可以选发送对象：单个用户、用户组、全量用户，或者直接上传 CSV 名单。

给单人发很简单。难点在“全量”：先查出所有设备的 token，再挨个发。所以我把用户组和全量推送改成了 FCM topic。发消息一下子变简单了，但绝大部分工作量转到了别的地方。

Topic 把工作转移了，并没有减少工作

用了 topic 之后，设备端只要订阅一个名字（比如 group_12），往这个名字发一条消息就能覆盖所有订阅者，FCM 负责分发。

但这样一来，“谁在这个组里”保存在我的数据库里，“谁订阅了这个 topic”保存在 FCM 那里。同一个事实变成了两份数据。这篇文章后面的所有内容，都是在解决这两份数据怎么保持一致。

最终的实现并不是彻底切过去，而是一个混合方案：

• 用户组和全量用户（仅限 FCM 设备）：一个用户组一个 topic，全量用户单独一个。
• 单人或 CSV 名单：还是按设备 token 发。
• 不用 FCM 的设备：走其他推送通道，按 token 发。所以“全量”实际上是一个 topic 发送加上一个 token 发送。

我试过给 CSV 名单用临时 topic，后来废掉了

我最初给 CSV 名单设计的是临时 topic 方案：把名单里的设备全部订阅到 temp_xxx，发完一次就标记废弃。这样能让 CSV 名单和用户组复用同一套逻辑。

这个方案只跑了六周左右。它有自己单独的表，后来重构时我全删了。提交记录里写的是不再需要这套独立的底层支持。在我查到的最新分支上，CSV 推送走的是 token。

而且这里面有个当时没查出来的 bug。定时清理任务只把数据库里的临时 topic 状态改成了停用，从来没去 FCM 帮设备退订，所以并没有真正清理掉这些订阅。

把订阅关系维护对

我加了一张表，记录每个 token 订阅各个 topic 的状态、最后确认时间、错误和重试信息。FCM 没有提供“这个 token 订阅了哪些 topic”的查询接口，不建这张表，我就答不上来“为什么这个用户没收到”。

用户组变动只是个开始。接下来的几个月里，报出来的 bug 全是同一个类型：两份数据脱节了。

• 用户关闭了推送，在他重新打开之前，设备不应该订阅任何 topic。
• 开关推送必须同步订阅状态，退订失败了需要重试。
• 删除成员时必须清理掉对应的 topic。
• 重复的设备注册可能会并发，所以我对注册逻辑加了串行化，给成员的 topic 更新加了队列。

速率限制，以及我实际写的代码

速率限制是写在配置里的，每个指标只用上限的 80%，留出余量：

const defaultLimits = {
  maxQPS: 3000,
  maxBatchSize: 1000,
  maxTopicsPerDevice: 2000,
  concurrentBatches: 3,
  rateLimitBuffer: 0.8,
};

“批处理”这个名字有点美化了后面的做法。FCM 的接口支持单次传入多个 token，但我的代码在每个批次内部依然是一个个 token 依次去订阅的。每次同时跑三个批次，轮次之间加了延迟：

async function subscribeBatches(requests: SubscribeRequest[], limits = defaultLimits) {
  const size = Math.floor(limits.maxBatchSize * limits.rateLimitBuffer);
  const qps = Math.floor(limits.maxQPS * limits.rateLimitBuffer);
  const delayMs = Math.ceil(1000 / (qps / size));
  const results: SubscribeResult[] = [];

  for (const window of chunk(requests, size * limits.concurrentBatches)) {
    const batchResults = await Promise.all(
      chunk(window, size).map(async (batch) => {
        const batchResult: SubscribeResult[] = [];
        for (const request of batch) {
          batchResult.push(await subscribeOne(request)); // one token per call
        }
        return batchResult;
      }),
    );
    results.push(...batchResults.flat());
    await sleep(delayMs);
  }
  return results;
}

这样能拿到每个 token 的单独结果，失败记录处理起来简单，但消耗的接口调用次数比实际需要的多得多。

在那个阶段，大批量的订阅任务还是放在进程内的普通数组队列里运行的，服务重启后不会保留。发送逻辑不一样：发送是丢给另一个推送服务处理的，用的是 Redis 上的 BullMQ 队列，支持重试。这两种队列的保证差别很大，但讨论时很容易混为一谈。

按语言拆分 topic

App 支持英文和中文，推送需要根据用户的语言发对应文案。一个 group_12 只能带一条消息，所以我按语言拆成了多个 topic：all_en、all_zh、group_12_en、group_12_zh。一条双语推送，在每个用户组下变成按语言各发一次。

老设备之前已经订阅了旧 topic，所以需要做一次迁移。我写了迁移脚本、回滚脚本，还设了一道必须手动输入确认才能在生产环境运行的拦截。

大概三个月后出了问题：设备可能会同时订阅两个语言的 topic。因为注册逻辑只管添加订阅，用户切语言时从来没去退订旧的。

修复办法是不再按“事件”去推算增量。在设备注册时，直接算出它“应该”订阅哪些 topic，跟数据库里的记录比对，把多余的退订掉：

async function reconcileDeviceTopics(deviceToken: string, userId: string, language: Lang) {
  const memberships = await db.groupMember.findMany({ where: { userId } });
  const desired = new Set([
    allTopic(language),
    ...memberships.map((m) => groupTopic(m.groupId, language)),
  ]);

  const existing = await db.topicSubscription.findMany({
    where: { deviceToken, status: { not: "UNSUBSCRIBED" } },
  });

  for (const { topicName } of existing) {
    if (!desired.has(topicName)) await unsubscribe(deviceToken, topicName);
  }
  await Promise.all([...desired].map((topicName) => subscribe(deviceToken, topicName)));
}

这是整个项目里我最想保留的做法。它不管设备是怎么变成当前状态的。不过这也没彻底解决所有问题。我写在后续计划里的项，依然包括对单 token 加锁（防止并发注册互相覆盖），以及把语言配置从用户维度改到设备维度（毕竟一个人两台手机可能用不同语言）。

“推送没收到”

在 Android 上，预览推送偶尔完全不展示，但几秒后发的正式推送就能正常收到。我搭了个环境重现：

1. 把一台 Android 测试机断网。
2. 先发预览推送，5 秒后发正式推送。
3. 大概 60 秒后把设备恢复联网。

推送服务显示两条都接受了，FCM 也显示两条都接受了，但设备上只弹出了正式推送。这支持我的推测：设备离线期间，FCM 替换了上一条未送达的消息。这只是推论，因为我看不到 FCM 内部的实现。报告里我也写了它解释不了的问题：之前遇到过的语言发错问题是另一个原因导致的。

这里面比较有用的习惯是把“没收到”拆成几个阶段去查：我们发出去没有、FCM 接受没有、设备收到没有。这也提醒了一件事：把设备订阅到 topic 上，并不保证设备能把每条消息都展示出来。

我的总结

发送从来不是难点。这次遇到的每个 bug，根源都在于：两个系统存了同一份数据，而代码是用事件去同步它们。最后真正管用的，是重新计算目标状态并纠正差额。