如何将Laravel中Elasticsearch滚动查询转换为MongoDB实现

问题描述 #

我需要把Laravel中基于Elasticsearch的滚动查询代码转换为MongoDB结构，但始终无法正确实现。请问如何将Elasticsearch中使用的scroll方法适配到MongoDB中，并在条件判断中正确处理该逻辑？

原Elasticsearch实现代码 #

private function comparedProduct()
{
    $this->info('Create for compared product');

   $comparedProductParam =
    [
        'meta_field' => 1,
        'with_scroll_id' => 1,
        'scroll' => $this->scrollDuration,
        'scroll_size' => $this->perPage,
        'index' => 'compared_product',
        'sort' => [["created_at" => ["order" => "ASC"]]],
    ];
    $scrollIdForComparedProduct = $this->esProductService->findBy($comparedProductParam);
    $skip = 0;
    $limit = $this->perPage; 

   // $comparedProducts = ComparedProduct::orderBy('created_at', 'asc')->skip($skip)->limit($limit)->get()->toArray();

    $comparedProducts = fillOnUndefined($scrollIdForComparedProduct, 'response', []);

    $this->info('compared product is : 1');

    $this->storeComparedProduct($comparedProducts);

    for ($i=2; $i <= 15; $i++)
    {
        $this->info('compared product is : '.$i);

        $scrollParam =
        [
            'scroll_id' => $scrollIdForComparedProduct['scroll_id'],
            'scroll' => $this->scrollDuration
        ];
        $response = $this->esProductService->scroll($scrollParam);
        if(!$response['response'])
        {
            continue;
        }

        $this->storeComparedProduct($response['response']);
    }

}

适配MongoDB的两种实现方案 #

方案一：使用MongoDB游标（推荐） #

MongoDB的游标机制与Elasticsearch的scroll原理一致，都是惰性加载数据，避免一次性加载全量数据到内存。Laravel的MongoDB Eloquent原生支持游标操作，直接替换ES的scroll逻辑即可：

private function comparedProduct()
{
    $this->info('Create for compared product');

    // 初始化游标：按created_at升序，惰性加载数据
    $cursor = ComparedProduct::orderBy('created_at', 'asc')->cursor();
    $batchSize = $this->perPage;
    $batchNumber = 1;

    // 处理第一批数据
    $comparedProducts = $cursor->take($batchSize)->toArray();
    $this->info('compared product is : ' . $batchNumber);
    $this->storeComparedProduct($comparedProducts);

    // 循环处理后续批次（最多15批）
    while ($batchNumber < 15 && $cursor->hasMoreRecords()) {
        $batchNumber++;
        $this->info('compared product is : ' . $batchNumber);
        
        $nextBatch = $cursor->take($batchSize)->toArray();
        // 空批次跳过，对应原ES逻辑的空响应判断
        if (empty($nextBatch)) {
            continue;
        }
        
        $this->storeComparedProduct($nextBatch);
    }
}

关键逻辑说明

• 游标初始化：cursor()创建惰性游标，仅在需要时从MongoDB拉取数据，和ES scroll的内存友好特性一致
• 分批获取：take($batchSize)每次从游标中提取指定数量的记录，对应ES的scroll_size
• 循环控制：hasMoreRecords()判断游标是否还有未读取的数据，同时限制最多15批，和原逻辑对齐
• 空数据处理：判断批次为空时跳过，匹配原ES代码中!$response['response']的逻辑

方案二：基于排序字段的分页查询（兼容无游标场景） #

如果无法使用游标，可以通过记录最后一条数据的排序字段值，实现类似ES滚动的分页查询，避免重复数据：

private function comparedProduct()
{
    $this->info('Create for compared product');

    $perPage = $this->perPage;
    $maxBatches = 15;
    $batchNumber = 1;
    $lastCreatedAt = null;

    while ($batchNumber <= $maxBatches) {
        $query = ComparedProduct::orderBy('created_at', 'asc');
        
        // 从上次最后一条记录的下一条开始查询，避免重复
        if ($lastCreatedAt) {
            $query->where('created_at', '>', $lastCreatedAt);
        }
        
        $comparedProducts = $query->take($perPage)->get()->toArray();
        $this->info('compared product is : ' . $batchNumber);
        
        // 无数据时终止循环
        if (empty($comparedProducts)) {
            break;
        }
        
        $this->storeComparedProduct($comparedProducts);
        
        // 更新最后一条记录的created_at，作为下一批查询的起点
        $lastCreatedAt = end($comparedProducts)['created_at'];
        $batchNumber++;
    }
}

关键逻辑说明

• 分页起点控制：通过where('created_at', '>', $lastCreatedAt)确保每次查询的是未处理过的数据，避免重复
• 循环终止条件：当批次为空时直接终止循环，比原逻辑的continue更高效
• 兼容性：适合无法使用游标的环境，逻辑更贴近传统分页，但实现了类似ES滚动的分批处理效果

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