如何在Firebase服务器端批量验证500个FCM Token？

FCM Instance ID官方API没有提供直接的批量验证接口，但可以通过以下几种高效方式处理500个Token的验证需求：

1. 并发异步请求 #

通过控制并发数发起异步请求，避免逐个同步请求的低效问题（建议设置10-20个并发，避免触发Google API的限流机制）。

Python异步脚本示例（使用aiohttp） #

import aiohttp
import asyncio

# 替换为你的认证Token和待验证的Token列表
FCM_AUTH_TOKEN = "ya29.ElqKBGN2Ri_Uz...HnS_uNreA"
TOKENS = ["nKctODamlM4:CKrh_PC8kIb7O...clJONHoA", "...", "剩余498个Token"]

async def validate_token(session, token):
    url = f"https://iid.googleapis.com/iid/info/{token}?details=true"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {FCM_AUTH_TOKEN}",
        "access_token_auth": "true"
    }
    try:
        async with session.get(url, headers=headers) as response:
            if response.status == 200:
                return (token, "有效", await response.json())
            elif response.status == 404:
                return (token, "无效", None)
            else:
                return (token, f"错误: {response.status}", None)
    except Exception as e:
        return (token, f"异常: {str(e)}", None)

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [validate_token(session, token) for token in TOKENS]
        results = await asyncio.gather(*tasks)
        # 将结果写入文件保存
        with open("fcm_token_validation_results.txt", "w") as f:
            for token, status, data in results:
                f.write(f"{token}: {status}\n")

if __name__ == "__main__":
    asyncio.run(main())

命令行并行请求示例（使用xargs+curl） #

假设你将所有Token存入tokens.txt（每行一个Token）：

cat tokens.txt | xargs -P 10 -I {} curl -s -o /dev/null -w "{}: %{http_code}\n" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer ya29.ElqKBGN2Ri_Uz...HnS_uNreA" \
  -H "access_token_auth: true" \
  "https://iid.googleapis.com/iid/info/{}?details=true"

2. 基于Firebase Admin SDK封装批量验证 #

利用Firebase Admin SDK的自动认证能力，不用手动管理Bearer Token，封装异步验证逻辑：

Python Admin SDK示例 #

import firebase_admin
from firebase_admin import credentials
import asyncio
import aiohttp

# 初始化Admin SDK（替换为你的服务账号密钥路径）
cred = credentials.Certificate("path/to/serviceAccountKey.json")
firebase_admin.initialize_app(cred)

TOKENS = ["nKctODamlM4:CKrh_PC8kIb7O...clJONHoA", "...", "剩余498个Token"]

async def validate_token(session, token):
    # 通过SDK自动获取有效访问令牌
    access_token = firebase_admin.credentials.ApplicationDefault().get_access_token().access_token
    url = f"https://iid.googleapis.com/iid/info/{token}?details=true"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {access_token}",
        "access_token_auth": "true"
    }
    try:
        async with session.get(url, headers=headers) as response:
            if response.status == 200:
                return (token, "有效")
            elif response.status == 404:
                return (token, "无效")
            else:
                return (token, f"错误: {response.status}")
    except Exception as e:
        return (token, f"异常: {str(e)}")

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [validate_token(session, token) for token in TOKENS]
        results = await asyncio.gather(*tasks)
        for res in results:
            print(res)

if __name__ == "__main__":
    asyncio.run(main())

关键注意事项 #

• 配额与限流：500个请求在FCM API配额范围内，但并发过高会触发429限流，建议严格控制并发数在10-20之间。
• 错误码解读：404表示Token已无效或被注销；401表示认证失败，需检查Token或SDK配置；429需添加重试机制或降低并发。
• 结果存储：将验证结果（有效/无效/错误）持久化到文件或数据库，方便后续清理无效Token。

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