.Net下如何并行批量或多实例处理MongoDB海量文档

针对你每天处理20万条MongoDB文档的需求，我来分享几个实用的优化思路和具体实现方案，帮你提升处理效率：

一、多任务并行处理批次的落地实现 #

你当前用游标分批加载的方案是MongoDB官方推荐的大数据量处理方式，核心优化点可以放在批次的并行处理上——不过要注意，游标本身不是线程安全的，所以我们要先把每个批次的文档拿到本地，再交给后台任务并行处理，同时要控制并发度避免资源过载。

具体代码示例（.NET环境） #

这里用SemaphoreSlim来限制并发任务数，防止同时处理太多批次导致内存、IO或MongoDB连接压力过大：

var options = new FindOptions 
{ 
    BatchSize = 2000, // 根据单文档大小调整，建议1000-5000之间
    NoCursorTimeout = true // 避免处理慢导致游标超时，用完务必关闭游标
};

// 控制并发度，一般设为CPU核心数的2倍即可
var concurrencySemaphore = new SemaphoreSlim(Environment.ProcessorCount * 2);
var processingTasks = new List<Task>();

using (var cursor = collection.Find(filter, options).ToCursor())
{
    while (cursor.MoveNext())
    {
        var batch = cursor.Current.ToList();
        await concurrencySemaphore.WaitAsync();

        // 将当前批次的处理逻辑交给后台任务
        processingTasks.Add(Task.Run(async () =>
        {
            try
            {
                // 建议用共享的文件流批量写入，避免频繁打开关闭文件
                using var writer = new StreamWriter("output.txt", append: true);
                foreach (var doc in batch)
                {
                    // 你的数据提取逻辑
                    var extractedRecord = ExtractRequiredData(doc);
                    await writer.WriteLineAsync(extractedRecord);
                }
            }
            finally
            {
                concurrencySemaphore.Release();
            }
        }));
    }

    // 等待所有批次处理完成
    await Task.WhenAll(processingTasks);
}

多实例部署的注意事项 #

如果想通过多实例进一步提升效率，需要拆分查询范围，避免多个实例重复处理同一份数据：

• 按_id范围拆分：比如把_id分成N段，每个实例处理其中一段
• 按分片键拆分：如果集合是分片集群，每个实例对应一个分片的查询
• 按业务字段拆分：比如按日期、用户ID等维度拆分查询条件

二、是否有比游标更好的加载方式？ #

游标本身就是MongoDB处理海量数据的最佳实践之一，但可以结合其他特性进一步优化：

1. 用异步流（IAsyncEnumerable）减少内存占用 #

如果你用的是.NET Core 3.0+，可以用ToAsyncEnumerable直接遍历游标，不需要一次性把整个批次加载到List中，适合单文档较大的场景：

var options = new FindOptions { BatchSize = 2000 };
var asyncCursor = await collection.Find(filter, options).ToCursorAsync();

await foreach (var doc in asyncCursor.ToAsyncEnumerable())
{
    // 异步处理单个文档，或攒成小批量后再写入文件
    var extractedData = ExtractRequiredData(doc);
    await AppendToFileAsync(extractedData);
}

2. 用聚合管道提前处理数据 #

如果你的提取逻辑可以用MongoDB的聚合操作实现，建议先在服务器端完成过滤、转换，减少客户端需要处理的数据量：

var pipeline = new BsonDocument[]
{
    BsonDocument.Parse("{ $match: " + filter.ToJson() + " }"),
    BsonDocument.Parse("{ $project: { requiredField1: 1, requiredField2: 1 } }") // 只返回需要的字段
};

var options = new AggregateOptions { BatchSize = 2000 };
using (var cursor = collection.Aggregate(pipeline, options).ToCursor())
{
    // 后续处理逻辑和游标方案一致
}

3. 批量写入优化 #

写入本地文件时，尽量避免单条写入，而是攒成一批后再批量写入，或者用带缓冲区的StreamWriter，能大幅提升IO效率。

最后几个小提示 #

• 确保你的查询filter用到了合适的索引，不然游标加载批次的速度会被拖慢
• BatchSize不要设置过大，否则单批次内存占用过高；也不要太小，会增加MongoDB的请求次数
• 开启NoCursorTimeout后，一定要确保游标被正确释放（用using包裹），避免服务器资源泄漏

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