Mongo C# Driver 聚合使用---深入浅出
聚合查询结构体系
我们都知道Mongo中聚合是由$match，$project等聚合项组成，所以在C# Driver中具有两种类型：聚合管道(PipelineDefinition)和聚合管道项(IPipelineStageDefinition) ，下面先来看一下聚合管道项的结构体系
IPipelineStageDefinition
IPipelineStageDefinition接口是聚合管道项的顶级接口，这个接口中只定义了一些获取输入类型和输出类型的简单的属性
public interface IPipelineStageDefinition
{
// 输入类型
Type InputType { get; }
// 获取管道操作名称
string OperatorName { get; }
// 输出类型
Type OutputType { get; }
// 获取一个IRenderedIRenderedPipelineStageDefinition
// IRenderedPipelineStageDefinition是一个Stage提供对象，下面会介绍
IRenderedPipelineStageDefinition Render(IBsonSerializer inputSerializer, IBsonSerializerRegistry serializerRegistry);
// 获取当前管道项的字符串格式，例：{ \"$match\" : { \"_id\" : \"402066782845407232\" } }
string ToString(IBsonSerializer inputSerializer, IBsonSerializerRegistry serializerRegistry);
} 这个接口具有一个PipelineStageDefinition派生类，这个类是一个抽象类，在这个抽象类中只多了两个隐式转换，
public abstract class PipelineStageDefinition<TInput, TOutput> : IPipelineStageDefinition
{
// 将一个BsonDocument对象转换为管道项
public static implicit operator PipelineStageDefinition<TInput, TOutput>(BsonDocument document);
// 将一个json字符串转换为管道项
public static implicit operator PipelineStageDefinition<TInput, TOutput>(string json);
} 用过C# Driver的朋友都应该知道我们使用Driver时经常使用这种隐式转换，例如经常使用FilterDefinition便可使用json字符串直接赋值,这也是Driver强大的地方。
FilterDefinition<BsonDocument> filter = "{_id:123}"; 其实这两个隐式转换如果翻源码就会看到直接创建了这个抽象类的实现类对象
public static implicit operator PipelineStageDefinition<TInput, TOutput>(BsonDocument document)
{
if (document == null)
return null;
return new BsonDocumentPipelineStageDefinition<TInput, TOutput>(document);
}
public static implicit operator PipelineStageDefinition<TInput, TOutput>(string json)
{
if (json == null)
return null;
return new JsonPipelineStageDefinition<TInput, TOutput>(json);
} 也就是说这个抽象类具有这么两个派生类BsonDocumentPipelineStageDefinition，JsonPipelineStageDefinition 这两个类型就是使用Bsondocument对象和json字符串进行实例化聚合管道项
PipelineStageDefinition其它派生类
如果仅仅使用，只使用上面那两个派生类即可，但实际上IPipelineStageDefinition的派生类还有两个：
DelegatedPipelineStageDefinition：由一个Func<IBsonSerializer委托创建的实例对象 SortPipelineStageDefinition：排序项的实例对象
其实这两个派生类在使用上根本不需要知道，它们的访问级别是internal,也就是说在使用时根本无法创建这两个派生类的实例对象，其实这两个类都是PipelineStageDefinition实例在调用Match() ，Project() ，Sort() 方法时进行内部创建的，这个下面再说
对于SortPipelineStageDefinition和DelegatedPipelineStageDefinition这两个派生类其实内部特别简单，但是却又扯到了另外一个类型
internal class SortPipelineStageDefinition<TInput> : PipelineStageDefinition<TInput, TInput>
{
public SortPipelineStageDefinition(SortDefinition<TInput> sort)
{
Sort = sort;
}
// 排序条件对象
public SortDefinition<TInput> Sort { get; private set; }
// 操作项名称
public override string OperatorName => "$sort";
public override RenderedPipelineStageDefinition<TInput> Render(IBsonSerializer<TInput> inputSerializer, IBsonSerializerRegistry serializerRegistry)
{
var renderedSort = Sort.Render(inputSerializer, serializerRegistry);
var document = new BsonDocument(OperatorName, renderedSort);
return new RenderedPipelineStageDefinition<TInput>(OperatorName, document, inputSerializer);
}
}
internal sealed class DelegatedPipelineStageDefinition<TInput, TOutput> : PipelineStageDefinition<TInput, TOutput>
{
// 委托缓存
private readonly Func<IBsonSerializer<TInput>, IBsonSerializerRegistry, RenderedPipelineStageDefinition<TOutput>> _renderer;
public DelegatedPipelineStageDefinition(string operatorName, Func<IBsonSerializer<TInput>, IBsonSerializerRegistry, RenderedPipelineStageDefinition<TOutput>> renderer)
{
_renderer = renderer;
}
// 获取RenderedPipelineStageDefinition 直接返回委托调用
public override RenderedPipelineStageDefinition<TOutput> Render(IBsonSerializer<TInput> inputSerializer, IBsonSerializerRegistry serializerRegistry)
{
return _renderer(inputSerializer, serializerRegistry);
}
} 通过上面代码可以看到这两个派生类型特别简单，感觉就是一个简单的代理，一切都指向于RenderedPipelineStageDefinition这个类型，也就是在真正执行聚合操作时可能使用的就是这个类型，这个在这先留一下悬念，因为RenderedPipelineStageDefinition这个类型还涉及到了整个聚合管道对象和执行操作，等到下面再讲解。
PipelineStageDefinitionBuilder
下面来说一下PipelineStageDefinitionBuilder这个类型，顾名思义，这是一个创建PipelineStageDefinition的类型，它是一个静态类，内部具有创建各种的使用方法，这个类型中方法特别多，也不一一细讲，只讲三个方法，也就是上面提到的Match() ，Project() ，Sort()
public static class PipelineStageDefinitionBuilder
{
// match
public static PipelineStageDefinition<TInput, TInput> Match<TInput>(
FilterDefinition<TInput> filter)
{
const string operatorName = "$match";
var stage = new DelegatedPipelineStageDefinition<TInput, TInput>(
operatorName,
(s, sr) => new RenderedPipelineStageDefinition<TInput>(operatorName, new BsonDocument(operatorName, filter.Render(s, sr)), s));
return stage;
}
// project
public static PipelineStageDefinition<TInput, TOutput> Project<TInput, TOutput>(ProjectionDefinition<TInput, TOutput> projection)
{
const string operatorName = "$project";
var stage = new DelegatedPipelineStageDefinition<TInput, TOutput>(
operatorName,
(s, sr) =>
{
var renderedProjection = projection.Render(s, sr);
BsonDocument document;
if (renderedProjection.Document == null)
document = new BsonDocument();
else
document = new BsonDocument(operatorName, renderedProjection.Document);
return new RenderedPipelineStageDefinition<TOutput>(operatorName, document, renderedProjection.ProjectionSerializer);
});
return stage;
}
// sort
public static PipelineStageDefinition<TInput, TInput> Sort<TInput>(
SortDefinition<TInput> sort)
{
return new SortPipelineStageDefinition<TInput>(sort);
}
} 上面就是这三个方法的源代码，三个方法分别使用FilterDefinition，ProjectionDefinition，SortDefinition实例创建PipelineStageDefinition对象，而所创建的也是后面讲的那两个派生类，这也验证了上面所说的两个类型的用途。
PipelineStageDefinition类总结
从上面一步步可以得知，Driver为我们提供了三种创建聚合项的办法，其实这三种也应用于driver的各种使用上
- BsonDocument创建
- json字符串创建
- 使用PipelineStageDefinitionBuilder进行创建
PipelineDefinition
说完管道项，下面就说一下整个聚合管道的操作类PipelineDefinition以及它的派生类
首先PipelineDefinition这个父级类型，它跟PipelineStageDefinition一样是一个抽象类型，并且和PipelineStageDefinition相同的是它也有一个Render方法和两个隐式转换，多了几个静态的创建方法，使得更具有扩展性
public abstract class PipelineDefinition<TInput, TOutput>
{
public abstract RenderedPipelineDefinition<TOutput> Render(IBsonSerializer<TInput> inputSerializer, IBsonSerializerRegistry serializerRegistry);
// 使用管道项集合创建一个PipelineStagePipelineDefinition实例对象
public static PipelineDefinition<TInput, TOutput> Create(
IEnumerable<IPipelineStageDefinition> stages,
IBsonSerializer<TOutput> outputSerializer = null)
{
return new PipelineStagePipelineDefinition<TInput, TOutput>(stages, outputSerializer);
}
// 使用BsonDocument集合创建一个BsonDocumentStagePipelineDefinition对象
public static PipelineDefinition<TInput, TOutput> Create(
IEnumerable<BsonDocument> stages,
IBsonSerializer<TOutput> outputSerializer = null)
{
return new BsonDocumentStagePipelineDefinition<TInput, TOutput>(stages, outputSerializer);
}
// 使用json字符串集合创建一个BsonDocumentStagePipelineDefinition对象
public static PipelineDefinition<TInput, TOutput> Create(
IEnumerable<string> stages,
IBsonSerializer<TOutput> outputSerializer = null)
{
return Create(stages?.Select(s => BsonDocument.Parse(s)), outputSerializer);
}
// 隐式转换
// 将IPipelineStageDefinition集合隐式转换为PipelineStagePipelineDefinition对象
public static implicit operator PipelineDefinition<TInput, TOutput>(List<IPipelineStageDefinition> stages)
{
return Create(stages);
}
// 将BsonDocument集合隐式转换为BsonDocumentStagePipelineDefinition对象
public static implicit operator PipelineDefinition<TInput, TOutput>(List<BsonDocument> stages)
{
return Create(stages);
}
}注：PipelineDefinition类中还封装了数组参数和其它内容，有兴趣的朋友可以自己去看看
上面类型可以看出PipelineDefinition做了很多封装，为了使用更加便捷。从上面也看到了两个派生类型：PipelineStagePipelineDefinition和BsonDocumentStagePipelineDefinition
其实PipelineDefinition派生类型一共有7个，我们能用到的是6个，我将这个7个类型分为：创建性，改变性和外部不可用性这三种，下面先来看看创建性
注：其实严格意义上是两种，外部不可用的派生类型属于创建性，外部不可用的派生类型也只是在特定情况下被内部用到。
创建性派PipelineDefinition
创建性有3个，其中两个就是上面基类中创建的两个派生类型，另外一个是EmptyPipelineDefinition，顾名思义这是一个空的管道，这个跟创建空条件那个是极其相似的( Builders)，
// EmptyPipelineDefinition
public sealed class EmptyPipelineDefinition<TInput> : PipelineDefinition<TInput, TInput>
{
public override IEnumerable<IPipelineStageDefinition> Stages => Enumerable.Empty<IPipelineStageDefinition>();
//
public override RenderedPipelineDefinition<TInput> Render(IBsonSerializer<TInput> inputSerializer, IBsonSerializerRegistry serializerRegistry)
{
var documents = Enumerable.Empty<BsonDocument>();
return new RenderedPipelineDefinition<TInput>(documents, _inputSerializer ?? inputSerializer);
}
}
// PipelineStagePipelineDefinition
public sealed class PipelineStagePipelineDefinition<TInput, TOutput> : PipelineDefinition<TInput, TOutput>
{
private readonly IList<IPipelineStageDefinition> _stages;
public PipelineStagePipelineDefinition(IEnumerable<IPipelineStageDefinition> stages, IBsonSerializer<TOutput> outputSerializer = null)
{
_stages = stages;
_outputSerializer = outputSerializer;
}
public override IEnumerable<IPipelineStageDefinition> Stages => _stages;
public override RenderedPipelineDefinition<TOutput> Render(IBsonSerializer<TInput> inputSerializer, IBsonSerializerRegistry serializerRegistry)
{
// 当前集合进行存储当前聚合管道所有聚合项的BsonDocument
var pipeline = new List<BsonDocument>();
IBsonSerializer currentSerializer = inputSerializer;
foreach (var stage in _stages)
{
// 获取每一个聚合项的RenderedPipelineDefinition
// 然后获取每个聚合项RenderedPipelineDefinition中的Bsondocument
var renderedStage = stage.Render(currentSerializer, serializerRegistry);
currentSerializer = renderedStage.OutputSerializer;
if (renderedStage.Document.ElementCount > 0)
{
pipeline.Add(renderedStage.Document);
}
}
return new RenderedPipelineDefinition<TOutput>(
pipeline,
_outputSerializer ?? (currentSerializer as IBsonSerializer<TOutput>) ?? serializerRegistry.GetSerializer<TOutput>());
}
// BsonDocumentStagePipelineDefinition
public sealed class BsonDocumentStagePipelineDefinition<TInput, TOutput> : PipelineDefinition<TInput, TOutput>
{
private readonly List<BsonDocument> _stages;
public BsonDocumentStagePipelineDefinition(IEnumerable<BsonDocument> stages, IBsonSerializer<TOutput> outputSerializer = null)
{
_stages = stages;
_outputSerializer = outputSerializer;
}
public override IBsonSerializer<TOutput> OutputSerializer => _outputSerializer;
public IList<BsonDocument> Documents
{
get { return _stages; }
}
// 获取当前聚合的所有聚合项
public override IEnumerable<IPipelineStageDefinition> Stages => _stages.Select(s => new BsonDocumentPipelineStageDefinition<TInput, TOutput>(s, _outputSerializer));
public override RenderedPipelineDefinition<TOutput> Render(IBsonSerializer<TInput> inputSerializer, IBsonSerializerRegistry serializerRegistry)
{
return new RenderedPipelineDefinition<TOutput>(
_stages,
_outputSerializer ?? (inputSerializer as IBsonSerializer<TOutput>) ?? serializerRegistry.GetSerializer<TOutput>());
}
} 上面是这个三个派生类型基本实现，基本上也都没什么特别的地方，而逻辑也是在Render()这个方法中，EmptyPipelineDefinition中创建了一个空的Bsondocument对象集合实例化的RenderedPipelineDefinition，而BsonDocumentStagePipelineDefinition和PipelineStagePipelineDefinition分别以传入的Bsondocument集合和从管道项对象中调用的Render()中获取Bsondocument集合。从这里可以得出2点
1.RenderedPipelineStageDefinition的作用是为了提供其内部的Bsondocument然后创建RenderedPipelineDefinition对象 2.RenderedPipelineStageDefinition和RenderedPipelineDefinition的关系就像BsonDocumentPipelineStageDefinition和BsonDocumentStagePipelineDefinition关系类似，一个对应管道项，一个对应管道
至此，一切的源头都指向了<span style="color:#009BDB">RenderedPipelineDefinition</span>这个类，但是这个类在下面再介绍，先来看一下改变性的<span style="color:#009BDB">PipelineDefinition</span>
改变性PipelineDefinition
为什么我叫它为改变性呢，因为它是在一个已有PipelineDefinition基础上进行的添加或者替换，下面来看看这三个派生类型
PrependedStagePipelineDefinition：在一个PipelineDefinition管道前面添加一个管道项 AppendedStagePipelineDefinition：在一个PipelineDefinition管道后面添加一个管道项 ReplaceOutputSerializerPipelineDefinition：替换一个PipelineDefinition的序列化对象类型
其实看到这三个派生类就知道其作用了，所以在这里也不进行详细介绍了，只贴出它们的构造方法，有兴趣的朋友可以翻阅源码
// PrependedStagePipelineDefinition
public sealed class PrependedStagePipelineDefinition<TInput, TIntermediate, TOutput> : PipelineDefinition<TInput, TOutput>
{
public PrependedStagePipelineDefinition(
PipelineStageDefinition<TInput, TIntermediate> stage,
PipelineDefinition<TIntermediate, TOutput> pipeline,
IBsonSerializer<TOutput> outputSerializer = null)
{
}
}
// AppendedStagePipelineDefinition
public sealed class AppendedStagePipelineDefinition<TInput, TIntermediate, TOutput> : PipelineDefinition<TInput, TOutput>
{
public AppendedStagePipelineDefinition(
PipelineDefinition<TInput, TIntermediate> pipeline,
PipelineStageDefinition<TIntermediate, TOutput> stage,
IBsonSerializer<TOutput> outputSerializer = null)
{
}
}
// ReplaceOutputSerializerPipelineDefinition
public sealed class ReplaceOutputSerializerPipelineDefinition<TInput, TIntermediate, TOutput> : PipelineDefinition<TInput, TOutput>
{
public ReplaceOutputSerializerPipelineDefinition(
PipelineDefinition<TInput, TIntermediate> pipeline,
IBsonSerializer<TOutput> outputSerializer = null)
{
}
}外部不可用派生类
这个外部不可用的派生类型是OptimizingPipelineDefinition ，按照翻译看起来像最优的管道，其实在执行操作时都会现将外部定义的PipelineDefinition转换为OptimizingPipelineDefinition 类型，首先先看看这个类型的定义
internal class OptimizingPipelineDefinition<TInput, TOutput> : PipelineDefinition<TInput, TOutput>
{
private readonly PipelineDefinition<TInput, TOutput> _wrapped;
public OptimizingPipelineDefinition(PipelineDefinition<TInput, TOutput> wrapped)
{
_wrapped = wrapped;
}
/// <inheritdoc />
public override IEnumerable<IPipelineStageDefinition> Stages => _wrapped.Stages;
public override RenderedPipelineDefinition<TOutput> Render(IBsonSerializer<TInput> inputSerializer, IBsonSerializerRegistry serializerRegistry)
{
var rendered = _wrapped.Render(inputSerializer, serializerRegistry);
if (rendered.Documents.Count > 1)
{
// 如果有可能，进行组合一下$match
var firstStage = rendered.Documents[0].GetElement(0);
var secondStage = rendered.Documents[1].GetElement(0);
if (firstStage.Name == "$match" && secondStage.Name == "$match")
{
var combinedFilter = Builders<BsonDocument>.Filter.And(
(BsonDocument)firstStage.Value,
(BsonDocument)secondStage.Value);
var combinedStage = new BsonDocument("$match", combinedFilter.Render(BsonDocumentSerializer.Instance, serializerRegistry));
rendered.Documents[0] = combinedStage;
rendered.Documents.RemoveAt(1);
}
}
return rendered;
}
} 可以看到只是在Render()代码中进行了一个轻微的优化操作，这个优化类是针对OfType情况进行优化的，唯一的使用地方是在FilteredMongoCollectionBase这个抽象类中，而这个抽象类的实现类是OfTypeMongoCollection
internal abstract class FilteredMongoCollectionBase<TDocument> : MongoCollectionBase<TDocument>, IFilteredMongoCollection<TDocument>
{
// 创建OptimizingPipelineDefinition
private PipelineDefinition<TDocument, TResult> CreateFilteredPipeline<TResult>(PipelineDefinition<TDocument, TResult> pipeline)
{
var filterStage = PipelineStageDefinitionBuilder.Match(_filter);
var filteredPipeline = new PrependedStagePipelineDefinition<TDocument, TDocument, TResult>(filterStage, pipeline);
return new OptimizingPipelineDefinition<TDocument, TResult>(filteredPipeline);
}
}
internal class OfTypeMongoCollection<TRootDocument, TDerivedDocument> : FilteredMongoCollectionBase<TDerivedDocument>
where TDerivedDocument : TRootDocument
{
}PipelineDefinitionBuilder类型
PipelineDefinitionBuilder类型是管道系列的一个帮助类，这个与PipelineStageDefinitionBuilder类相似，但又不尽相同，PipelineDefinitionBuilder中定义的都是PipelineDefinition对象的扩展方法，定义了一系列方便的方法
public static class PipelineDefinitionBuilder
{
// $match
public static PipelineDefinition<TInput, TOutput> Match<TInput, TOutput>(
this PipelineDefinition<TInput, TOutput> pipeline,
FilterDefinition<TOutput> filter)
{
return pipeline.AppendStage(PipelineStageDefinitionBuilder.Match(filter));
}
// $project
public static PipelineDefinition<TInput, TOutput> Project<TInput, TIntermediate, TOutput>(
this PipelineDefinition<TInput, TIntermediate> pipeline,
ProjectionDefinition<TIntermediate, TOutput> projection)
{
return pipeline.AppendStage(PipelineStageDefinitionBuilder.Project(projection));
}
// AppendStage
public static PipelineDefinition<TInput, TOutput> AppendStage<TInput, TIntermediate, TOutput> (
this PipelineDefinition<TInput, TIntermediate> pipeline,
PipelineStageDefinition<TIntermediate, TOutput> stage,
IBsonSerializer<TOutput> outputSerializer = null)
{
return new AppendedStagePipelineDefinition<TInput, TIntermediate, TOutput>(pipeline, stage, outputSerializer);
}
public static PipelineDefinition<TInput, TOutput> As<TInput, TIntermediate, TOutput>(
this PipelineDefinition<TInput, TIntermediate> pipeline,
IBsonSerializer<TOutput> outputSerializer = null)
{
return new ReplaceOutputSerializerPipelineDefinition<TInput, TIntermediate, TOutput>(pipeline, outputSerializer);
}
} 其实可以看出从上面几个个方法可以看出其本质还是使用AppendedStagePipelineDefinition和ReplaceOutputSerializerPipelineDefinition。Match()和Project()都是调用了AppendStage(),而这个方法是创建了一个新的AppendedStagePipelineDefinition对象返回。而As()也是创建了一个新的ReplaceOutputSerializerPipelineDefinition返回。其本质没变，但是可以使得整个driver多了扩展性，更加方便了使用。有的聚合项像$addFields并没有封装方法，可能使用率不大，所以并没有封装，像这样的直接就调用AppendStage()即可
PipelineDefinition类总结
通过上面介绍其实可以看出来了，Mongo的C# Driver中聚合操作使用起来特别方便，使用时先创建聚合项对象再创建聚合管道对象还是直接创建聚合管道对象或者直接使用隐式转换都可以。其实不止聚合，C# Driver中各个操作基本都是如此，使用起来都特别方便，既然创建聚合管道实例的方法特别多，所以在这也就不一一列出，只简单的列出几个
1.先实例化聚合项，再实例化聚合管道对象
2.直接使用隐式转换进行创建聚合管道对象
3.使用扩展方法进行创建
RenderedPipelineStageDefinition和RenderedPipelineDefinition介绍
下面我们来说说RenderedPipelineStageDefinition和RenderedPipelineDefinition这两个类，这两个类叫做聚合项和聚合管道的提供者，它们真正提供了聚合的语句。上面已经简单说过，它们分别是聚合项实例和聚合管道实例创建的，并且在PipelineStagePipelineDefinition中也可以看到RenderedPipelineDefinition是根据RenderedPipelineStageDefinition内部BsonDocument进行实例化的，下面先来看一看这两个类型的内部结构
// RenderedPipelineStageDefinition
public class RenderedPipelineStageDefinition<TOutput> : IRenderedPipelineStageDefinition
{
private string _operatorName;
private BsonDocument _document;
private IBsonSerializer<TOutput> _outputSerializer;
public RenderedPipelineStageDefinition(string operatorName, BsonDocument document, IBsonSerializer<TOutput> outputSerializer)
{
_operatorName = Ensure.IsNotNull(operatorName, nameof(operatorName));
_document = Ensure.IsNotNull(document, nameof(document));
_outputSerializer = Ensure.IsNotNull(outputSerializer, nameof(outputSerializer));
}
public BsonDocument Document
{
get { return _document; }
}
public IBsonSerializer<TOutput> OutputSerializer
{
get { return _outputSerializer; }
}
public string OperatorName
{
get { return _operatorName; }
}
IBsonSerializer IRenderedPipelineStageDefinition.OutputSerializer
{
get { return _outputSerializer; }
}
}
// RenderedPipelineDefinition
public class RenderedPipelineDefinition<TOutput>
{
private List<BsonDocument> _documents;
private IBsonSerializer<TOutput> _outputSerializer;
public RenderedPipelineDefinition(IEnumerable<BsonDocument> documents, IBsonSerializer<TOutput> outputSerializer)
{
_documents = Ensure.IsNotNull(documents, nameof(documents)).ToList();
_outputSerializer = Ensure.IsNotNull(outputSerializer, nameof(outputSerializer));
}
public IList<BsonDocument> Documents
{
get { return _documents; }
}
public IBsonSerializer<TOutput> OutputSerializer
{
get { return _outputSerializer; }
}
} 可以看到其实最重要是就是内部的BsonDocument这个属性，那么这个属性里面是什么呢，我们先来看一下
可以看出BsonDocument其实存放就是一个聚合项的json字符串，也就是
注：这个Render()是以序列化器类型实例和序列化注册实例进行序列化为字符串的
然后我来验证聚合的最后执行操作，也就是RenderedPipelineDefinition的作用，这个操作是在MongoCollectionImpl中，从下面代码可以看出，使用Render()方法获取聚合管道的真实语句。然后由此语句执行，由此可以看出其实一切的PipelineDefinition对象最终都是生成RenderedPipelineDefinition对象，这个对象携带着执行语句的json字符串形式。
internal sealed class MongoCollectionImpl<TDocument> : MongoCollectionBase<TDocument>
{
public override IAsyncCursor<TResult> Aggregate<TResult>(IClientSessionHandle session, PipelineDefinition<TDocument, TResult> pipeline, AggregateOptions options, CancellationToken cancellationToken = default(CancellationToken))
{
// 获取当前聚合管道对象的语句
var renderedPipeline = pipeline.Render(_documentSerializer, _settings.SerializerRegistry);
options = options ?? new AggregateOptions();
var last = renderedPipeline.Documents.LastOrDefault();
if (last != null && last.GetElement(0).Name == "$out")
{
var aggregateOperation = CreateAggregateToCollectionOperation(renderedPipeline, options);
ExecuteWriteOperation(session, aggregateOperation, cancellationToken);
var findOperation = CreateAggregateToCollectionFindOperation(last, renderedPipeline.OutputSerializer, options);
var forkedSession = session.Fork();
var deferredCursor = new DeferredAsyncCursor<TResult>(
() => forkedSession.Dispose(),
ct => ExecuteReadOperation(forkedSession, findOperation, ReadPreference.Primary, ct),
ct => ExecuteReadOperationAsync(forkedSession, findOperation, ReadPreference.Primary, ct));
return deferredCursor;
}
else
{
var aggregateOperation = CreateAggregateOperation(renderedPipeline, options);
return ExecuteReadOperation(session, aggregateOperation, cancellationToken);
}
}
}聚合操作的执行方法
上面说了整个聚合管道类的体系，下面说一下最后调用的执行方法
执行方法调用的是IMongoCollection对象的Aggregate()方法，这个方法在IMongoCollection类中具有两个重载，都是需要PipelineDefinition为参数的。
在这个方法中还有一个AggregateOptions参数。这个类是执行聚合的一些选择操作。比如是否使用游标，如果内存不足情况下是否允许使用磁盘等等。。
IAsyncCursor<TResult> Aggregate<TResult>(PipelineDefinition<TDocument, TResult> pipeline, AggregateOptions options = null, CancellationToken cancellationToken = default(CancellationToken));Aggregate()方法会返回一个IAsyncCursor实例，这个对象代表一个游标。
其实在IMongoCollectionExtensions这个扩展类中还具有Aggregate()方法，这个方法也算是另外一种用法。因为这个方法参数并没有PipelineDefinition对象，并且返回类型也不再是IAsyncCursor，而是一个IAggregateFluent类型。IAggregateFluent类型具有一系列方法
public static IAggregateFluent<TDocument> Aggregate<TDocument>(this IMongoCollection<TDocument> collection, AggregateOptions options = null);