基于Node.js+Express的FCM批量推送全用户通知方案咨询

我之前也碰到过类似的大规模FCM推送场景，单次1000个令牌的限制确实是个坎儿，结合Kue来做异步批量处理是非常合适的方案，下面给你一套完整的实现思路和代码示例，亲测好用：

核心思路：令牌分批 + 异步队列 #

FCM的sendMulticast接口单次最多支持1000个设备令牌，所以核心逻辑就是：

1. 从数据库分页查询所有有效的FCM令牌
2. 每1000个令牌分为一批，将每批的推送任务丢进Kue队列
3. 由队列异步处理每一批推送，同时处理失败重试、限流等问题

具体实现步骤（基于Node.js） #

1. 初始化依赖和客户端 #

首先确保你已经安装了kue、firebase-admin和对应的数据库驱动，然后初始化队列和FCM客户端：

const kue = require('kue');
const { initializeApp } = require('firebase-admin/app');
const { getMessaging } = require('firebase-admin/messaging');

// 初始化Firebase Admin（替换成你的服务账号密钥）
const firebaseApp = initializeApp({
  credential: require('./serviceAccountKey.json')
});
const messaging = getMessaging(firebaseApp);

// 初始化Kue队列（依赖Redis，确保本地或服务器上的Redis已启动）
const queue = kue.createQueue();

2. 编写队列任务处理器 #

定义一个处理单批令牌推送的任务函数，包含失败重试、日志记录等关键逻辑：

// 处理FCM推送任务，并发数设为5（可根据FCM配额调整）
queue.process('fcm-push', 5, async (job, done) => {
  const { tokens, notification } = job.data;
  
  try {
    // 调用FCM批量推送接口
    const response = await messaging.sendMulticast({
      tokens: tokens,
      notification: notification
    });

    // 收集推送失败的令牌，后续重试
    const failedTokens = [];
    response.responses.forEach((resp, index) => {
      if (!resp.success) {
        failedTokens.push(tokens[index]);
        console.error(`推送失败至令牌 ${tokens[index]}: ${resp.error.message}`);
      }
    });

    // 对失败令牌进行重试（最多3次）
    if (failedTokens.length > 0 && job.attempts < 3) {
      queue.create('fcm-push', {
        tokens: failedTokens,
        notification: notification
      }).attempts(3).save(err => {
        if (err) console.error('失败令牌重新入队失败:', err);
      });
    }

    done();
  } catch (error) {
    console.error('推送任务执行失败:', error);
    // 标记任务失败，Kue会自动触发重试
    done(error);
  }
});

3. 分批获取令牌并加入队列 #

写一个函数，从数据库分页拉取令牌，每批1000个，然后创建队列任务：

async function enqueueFullPush(notification) {
  const batchSize = 1000;
  let offset = 0;

  while (true) {
    // 替换成你的数据库查询逻辑，分页获取有效FCM令牌
    const userRecords = await yourDatabase.query(`
      SELECT fcm_token FROM users 
      WHERE fcm_token IS NOT NULL AND fcm_token != ''
      LIMIT ${batchSize} OFFSET ${offset}
    `);

    const tokens = userRecords.map(record => record.fcm_token);
    if (tokens.length === 0) break;

    // 将当前批次加入队列
    queue.create('fcm-push', {
      tokens: tokens,
      notification: notification
    }).attempts(3).save(err => {
      if (err) console.error(`批次 ${offset/batchSize + 1} 入队失败:`, err);
    });

    offset += batchSize;
    console.log(`已加入批次 ${offset/batchSize}，共 ${tokens.length} 个令牌`);
  }
}

4. 管理员触发推送的API接口 #

在你的管理后台API中添加一个接口，让管理员发起全量推送：

// 假设使用Express框架
const express = require('express');
const app = express();
app.use(express.json());

app.post('/api/admin/broadcast-push', async (req, res) => {
  const { title, body } = req.body;
  
  if (!title || !body) {
    return res.status(400).json({ error: '推送标题和内容不能为空' });
  }

  try {
    await enqueueFullPush({
      title: title,
      body: body
    });
    res.json({ message: '全量推送任务已加入队列，将异步处理' });
  } catch (error) {
    console.error('触发全量推送失败:', error);
    res.status(500).json({ error: '服务器内部错误，推送任务未触发' });
  }
});

app.listen(3001, () => console.log('管理后台API启动'));

关键优化建议 #

• 令牌清理：定期清理数据库中无效的FCM令牌（比如推送返回InvalidRegistration的令牌），减少无效请求
• 监控与日志：启动Kue的UI面板（kue.app.listen(3002)），可以直观查看任务的成功/失败/等待状态；同时将推送日志持久化到数据库，方便后续排查问题
• 限流控制：根据你的FCM配额调整队列的并发数，避免触发FCM的请求限流（FCM对不同等级的项目有不同的每分钟请求限额）
• 重试策略：对于多次重试仍失败的令牌，建议单独存储，后续人工排查原因（比如用户卸载应用、令牌过期等）

备选方案 #

如果Kue不符合你的技术栈或需求，也可以考虑用Bull（功能更丰富的Node.js队列库），或者自己基于数据库实现简单的任务队列，但成熟的队列库能帮你省掉很多任务持久化、重试、并发控制的麻烦。

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