MongoDB 游标分批查询与聚合

示例说明

查询结果集大时一次性拉取会同时打爆网络与内存——游标按批交付：服务器保留结果位置，客户端每批拉 batch_size 个文档，内存峰值与集合总量解耦。advance 推进到下一批，deserialize_current 反序列化当前批的全部文档。聚合管道在服务端完成分组求和，max_time 给慢聚合设硬性上限。

技术要点

• batch_size 控制单批网络传输量
• advance 与 deserialize_current 分离推进与解码
• 聚合的 max_time 限制服务端耗时

代码解读

find 查询按状态过滤、时间倒序、限量两千，游标循环里累计条数与金额；aggregate 的三段管道先过滤再按区域分组求和排序。两处游标消费形态一致——批粒度的 advance 让取消检查有了自然插入点。最后的 shutdown 优雅关闭连接池。

使用建议

游标长时间闲置会被服务端回收，消费间隙大要配会话续期。聚合输出小可直接 to_vec。生产慢查询配 hint 与索引评审。

代码示例

use mongodb::{
    bson::{doc, Document},
    options::{AggregateOptions, FindOptions},
    Client,
};
use std::time::Duration;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = Client::with_uri_str("mongodb://localhost:27017").await?;
    let orders = client.database("shop").collection::<Document>("orders");

    // 游标按批拉取：batch_size 控制单次往返的文档数
    let find_options = FindOptions::builder()
        .batch_size(200)
        .limit(2000)
        .sort(doc! { "created_at": -1 })
        .build();

    let mut cursor = orders
        .find(doc! { "status": "paid" })
        .with_options(find_options)
        .await?;

    let mut total_value: i64 = 0;
    let mut count = 0usize;
    while cursor.advance().await? {
        for doc in cursor.deserialize_current()? {
            count += 1;
            total_value += doc.get_i64("total_cents").unwrap_or(0);
        }
    }
    println!("find: {count} orders worth {total_value} cents");

    // 聚合管道：服务端分组统计，游标同样分批消费
    let pipeline = vec![
        doc! { "$match": doc! { "status": "paid" } },
        doc! {
            "$group": {
                "_id": "$region",
                "revenue": doc! { "$sum": "$total_cents" },
                "orders": doc! { "$sum": 1 },
            }
        },
        doc! { "$sort": doc! { "revenue": -1 } },
    ];

    let aggregate_options = AggregateOptions::builder()
        .max_time(Duration::from_secs(10))
        .build();

    let mut agg_cursor = orders
        .aggregate(pipeline)
        .with_options(aggregate_options)
        .await?;

    while agg_cursor.advance().await? {
        for doc in agg_cursor.deserialize_current()? {
            println!(
                "region {:?}: {} orders, {} cents",
                doc.get_str("_id").unwrap_or("?"),
                doc.get_i64("orders").unwrap_or(0),
                doc.get_i64("revenue").unwrap_or(0)
            );
        }
    }

    client.shutdown().await;
    Ok(())
}