如何将Google Cloud Task按Firebase数据拆分子任务规避内存限制

问题背景 #

我有一个Google Cloud Task，用于从Firebase数据库获取所有企业数据，遍历每个企业调用更新子任务。但企业数量增长后，这种循环方式触发了内存限制问题。当前代码如下：

$router->get('companies', function () use ($router) {

    $slackDataHelpersService = new \App\Services\SlackDataHelpersService();
    $companiesDocuments = $slackDataHelpersService->getCompanies();

    foreach ($companiesDocuments->documents() as $document) {
        $cid = $document->id();
        createTask('companies', 'updateCompany', "{$cid}");
    }

    return res(200, 'Task done');
});

我尝试过分页查询但未成功（以成员数据为例的尝试代码如下），想知道如何将企业文档拆分为多个批次，让每个任务处理100条文档而非全部：

$router->get('test2', function () use ($router) {

    $db = app('firebase.firestore')->database();

    $membersRef = $db->collection('companies')->document('slack-T01L7H2NDPB')->collection('members');
    $query = $membersRef->orderBy('created', 'desc')->limit(10);

    $perPage = 10;
    $batchCount = 10;
    $lastCreated = null;

    while ($batchCount == $perPage) {

        $loopQuery = clone $query;
        if ($lastCreated != null) {
            $loopQuery->startAfter($lastCreated);
        }
        $docs = $loopQuery->documents();
        $docsRows = $docs->rows();
        $batchCount = count($docsRows);

        if ($batchCount > 1) {
            $lastCreated = $docsRows[$batchCount - 1];
        }
        echo $lastCreated['created'];
        //createTasksByDocs($docs);
    }
    //return res(200, 'Task done');
});

解决方案 #

核心思路 #

利用Firestore的游标分页特性，每次获取固定数量的文档，为每个批次创建独立的处理任务，避免一次性加载全量数据占用内存。关键是通过startAfter基于已索引的排序字段实现精准分页。

修正后的主任务代码 #

$router->get('companies', function () use ($router) {
    $db = app('firebase.firestore')->database();
    $companiesRef = $db->collection('companies');
    
    // 定义每批次处理的文档数量
    $batchSize = 100;
    // 选择已创建索引的排序字段（例如文档创建时间或ID，需确保Firestore已建对应索引）
    $orderByField = 'created'; 

    $lastDocument = null;
    do {
        // 构建分页查询
        $query = $companiesRef->orderBy($orderByField)->limit($batchSize);
        if ($lastDocument) {
            $query->startAfter($lastDocument);
        }
        
        $documents = $query->documents();
        $docsArray = $documents->rows();
        $currentBatchCount = count($docsArray);
        
        if ($currentBatchCount > 0) {
            // 提取当前批次的企业ID列表
            $companyIds = array_map(function($doc) {
                return $doc->id();
            }, $docsArray);
            
            // 创建批次处理任务，将ID列表以JSON格式传递
            createBatchTask('companies', 'updateCompaniesBatch', json_encode($companyIds));
            
            // 更新游标，指向当前批次最后一个文档
            $lastDocument = end($docsArray);
        }
    } while ($currentBatchCount === $batchSize);

    return res(200, 'Batch tasks initiated');
});

批次处理子任务实现 #

// 批次处理子任务逻辑
function updateCompaniesBatch($companyIdsJson) {
    $companyIds = json_decode($companyIdsJson, true);
    $slackDataHelpersService = new \App\Services\SlackDataHelpersService();
    
    foreach ($companyIds as $cid) {
        // 复用原有单个企业更新逻辑
        updateCompany($cid);
    }
}

// 原有单个企业更新逻辑（可直接复用）
function updateCompany($cid) {
    $slackDataHelpersService = new \App\Services\SlackDataHelpersService();
    $company = $slackDataHelpersService->getCompanyById($cid);
    // ... 执行你的企业更新操作
}

关键注意事项 #

• 索引要求：排序字段必须在Firebase控制台创建对应索引，否则分页查询会失败。
• 游标准确性：startAfter需传入完整的文档对象，而非单个字段值，避免因字段重复导致数据遗漏或重复处理。
• 参数轻量化：批次任务传递企业ID列表而非完整文档数据，减少任务负载与内存占用。
• 异常防护：建议在批次任务中添加异常捕获逻辑，避免单个企业处理失败导致整个批次任务中断。

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