最近工作中需要用到MongoDB的事务操作，因此参考了一些资料封装了一个小的组件，提供基础的CRUD Repository基类 和 UnitOfWork工作单元模式。今天，就来简单介绍一下这个小组件。
关于MongoDB的事务
MongoDB在4.2版本开始全面支持了多文档事务，至今已过了四年了，虽然我们可能没有在项目中用MongoDB来替代传统关系型数据库如MySQL/SQL Server，但是不能否认MongoDB已经在事务能力上愈发成熟了。
在MongoDB中，所谓的事务主要指的是多个文档的事务，其使用方式和传统关系型数据库差不多。但我们需要注意的是：多文档事务只能应用在副本集 或 mongos 节点上。如果你只是一个单点的mongo实例，是无法进行多文档事务实践的。
画外音：如果你对MongoDB感兴趣，不妨看看我的这个系列博客：《MongoDB入门到实践学习之旅》
那么，如何快速进行事务操作呢？
在Mongo Shell中进行事务
下面演示了如何通过Mongo Shell来进行一个多文档操作的事务提交：
var session = db.getMongo().startSession();
session.startTransaction({readConcern: { level: 'majority' },writeConcern: { w: 'majority' }});
var coll1 = session.getDatabase('students').getCollection('teams');
coll1.update({name: 'yzw-football-team'}, {$set: {members: 20}});
var coll2 = session.getDatabase('students').getCollection('records');
coll1.update({name: 'Edison'}, {$set: {gender: 'Female'}});
// 成功提交事务
session.commitTransaction();
// 失败事务回滚
session.abortTransaction();
在.NET应用中进行事务
下面展示了在.NET应用中通过MongoDB Driver来进行事务的示例：
using (var clientSession = mongoClient.StartSession())
{
try
{
var contacts = clientSession.Client.GetDatabase("testDB").GetCollection<Contact>("contacts");
contacts.ReplaceOne(contact => contact.Id == "1234455", contact);
var books = clientSession.Client.GetDatabase("testDB").GetCollection<Book>("books");
books.DeleteOne(book => book.Id == "1234455");
clientSession.CommitTransaction();
}
catch (Exception ex)
{
// to do some logging
clientSession.AbortTransaction();
}
}
在大部分的实际应用中，我们通常都习惯使用数据仓储（Repository）的模式来进行CRUD，同时也习惯用工作单元（UnitOfWork）模式来进行协调多个Repository进行事务提交。那么，如何在自己的项目中实现这个呢？
参考了一些资料后，自己实现了一个基础小组件，暂且叫它：EDT.MongoProxy吧，我们来看看它是如何实现的。
单例的MongoClient
基于MongoDB的最佳时间，对于MongoClient最好设置为单例注入，因为在MongoDB.Driver中MongoClient已经被设计为线程安全可以被多线程共享，这样可还以避免反复实例化MongoClient带来的开销，避免在极端情况下的性能低下。
这里暂且设计一个MongoDbConnection类，用于包裹这个MongoClient，然后将其以单例模式注入IoC容器中。
public class MongoDbConnection : IMongoDbConnection
{
public IMongoClient DatabaseClient { get; }
public string DatabaseName { get; }
public MongoDbConnection(MongoDatabaseConfigs configs, IConfiguration configuration)
{
DatabaseClient = new MongoClient(configs.GetMongoClientSettings(configuration));
DatabaseName = configs.DatabaseName;
}
}
其中，这个MongoDatabaseConfigs类主要是获取appsettings中的配置，用以生成MongoClient的对应Settings。
/** Config Example
"MongoDatabaseConfigs": {
"Servers": "xxx01.edisontalk.net,xxx02.edisontalk.net,xxx03.edisontalk.net",
"Port": 27017,
"ReplicaSetName": "edt-replica",
"DatabaseName": "EDT_Practices",
"AuthDatabaseName": "admin",
"ApplicationName": "Todo",
"UserName": "service_testdev",
"Password": "xxxxxxxxxxxxxxxxxxxxxxxx",
"UseTLS": true,
"AllowInsecureTLS": true,
"SslCertificatePath": "/etc/pki/tls/certs/EDT_CA.cer",
"UseEncryption": true
}
**/
public class MongoDatabaseConfigs
{
private const string DEFAULT_AUTH_DB = "admin"; // Default AuthDB: admin
public string Servers { get; set; }
public int Port { get; set; } = 27017; // Default Port: 27017
public string ReplicaSetName { get; set; }
public string DatabaseName { get; set; }
public string DefaultCollectionName { get; set; } = string.Empty;
public string ApplicationName { get; set; }
public string UserName { get; set; }
public string Password { get; set; }
public string AuthDatabaseName { get; set; } = DEFAULT_AUTH_DB; // Default AuthDB: admin
public string CustomProperties { get; set; } = string.Empty;
public bool UseTLS { get; set; } = false;
public bool AllowInsecureTLS { get; set; } = true;
public string SslCertificatePath { get; set; } = string.Empty;
public bool StoreCertificateInKeyStore { get; set; } = false;
public MongoClientSettings GetMongoClientSettings(IConfiguration configuration = null)
{
if (string.IsNullOrWhiteSpace(Servers))
throw new ArgumentNullException("Mongo Servers Configuration is Missing!");
if (string.IsNullOrWhiteSpace(UserName) || string.IsNullOrWhiteSpace(Password))
throw new ArgumentNullException("Mongo Account Configuration is Missing!");
// Base Configuration
MongoClientSettings settings = new MongoClientSettings
{
ApplicationName = ApplicationName,
ReplicaSetName = ReplicaSetName
};
// Credential
settings.Credential = MongoCredential.CreateCredential(AuthDatabaseName, UserName, Password);
// Servers
var mongoServers = Servers.Split(",", StringSplitOptions.RemoveEmptyEntries).ToList();
if (mongoServers.Count == 1) // Standalone
{
settings.Server = new MongoServerAddress(mongoServers.First(), Port);
settings.DirectConnection = true;
}
if (mongoServers.Count > 1) // Cluster
{
var mongoServerAddresses = new List<MongoServerAddress>();
foreach (var mongoServer in mongoServers)
{
var mongoServerAddress = new MongoServerAddress(mongoServer, Port);
mongoServerAddresses.Add(mongoServerAddress);
}
settings.Servers = mongoServerAddresses;
settings.DirectConnection = false;
}
// SSL
if (UseTLS)
{
settings.UseTls = true;
settings.AllowInsecureTls = AllowInsecureTLS;
if (string.IsNullOrWhiteSpace(SslCertificatePath))
throw new ArgumentNullException("SslCertificatePath is Missing!");
if (StoreCertificateInKeyStore)
{
var localTrustStore = new X509Store(StoreName.Root);
var certificateCollection = new X509Certificate2Collection();
certificateCollection.Import(SslCertificatePath);
try
{
localTrustStore.Open(OpenFlags.ReadWrite);
localTrustStore.AddRange(certificateCollection);
}
catch (Exception ex)
{
throw;
}
finally
{
localTrustStore.Close();
}
}
var certs = new List<X509Certificate> { new X509Certificate2(SslCertificatePath) };
settings.SslSettings = new SslSettings();
settings.SslSettings.ClientCertificates = certs;
settings.SslSettings.EnabledSslProtocols = System.Security.Authentication.SslProtocols.Tls13;
}
return settings;
}
}
核心部分：MongoDbContext
这里我们主要仿照DbContext的设计，设计一个MongoDbContext，它从IoC容器中获取到单例的MongoClient，封装了事务的开启和提交，简化了应用代码的编写。
public class MongoDbContext : IMongoDbContext
{
private readonly IMongoDatabase _database;
private readonly IMongoClient _mongoClient;
private readonly IList<Func<IClientSessionHandle, Task>> _commands
= new List<Func<IClientSessionHandle, Task>>();
public MongoDbContext(IMongoDbConnection dbClient)
{
_mongoClient = dbClient.DatabaseClient;
_database = _mongoClient.GetDatabase(dbClient.DatabaseName);
}
public void AddCommand(Func<IClientSessionHandle, Task> func)
{
_commands.Add(func);
}
public async Task AddCommandAsync(Func<IClientSessionHandle, Task> func)
{
_commands.Add(func);
await Task.CompletedTask;
}
/// <summary>
/// NOTES: Only works in Cluster mode
/// </summary>
public int Commit(IClientSessionHandle session)
{
try
{
session.StartTransaction();
foreach (var command in _commands)
{
command(session);
}
session.CommitTransaction();
return _commands.Count;
}
catch (Exception ex)
{
session.AbortTransaction();
return 0;
}
finally
{
_commands.Clear();
}
}
/// <summary>
/// NOTES: Only works in Cluster mode
/// </summary>
public async Task<int> CommitAsync(IClientSessionHandle session)
{
try
{
session.StartTransaction();
foreach (var command in _commands)
{
await command(session);
}
await session.CommitTransactionAsync();
return _commands.Count;
}
catch (Exception ex)
{
await session.AbortTransactionAsync();
return 0;
}
finally
{
_commands.Clear();
}
}
public IClientSessionHandle StartSession()
{
var session = _mongoClient.StartSession();
return session;
}
public async Task<IClientSessionHandle> StartSessionAsync()
{
var session = await _mongoClient.StartSessionAsync();
return session;
}
public IMongoCollection<T> GetCollection<T>(string name)
{
return _database.GetCollection<T>(name);
}
public void Dispose()
{
GC.SuppressFinalize(this);
}
}
数据仓储：MongoRepositoryBase
在实际项目中，我们都希望有一个基础的RepositoryBase类，将CRUD的方法都封装了，我们实际中就只需要创建一个对应的Repository集成这个RepositoryBase就行了，无需再重复编写CRUD的方法。那么，也就有了这个MongoRepositoryBase类：
public class MongoRepositoryBase<TEntity> : IMongoRepositoryBase<TEntity>
where TEntity : MongoEntityBase, new()
{
protected readonly IMongoDbContext _dbContext;
protected readonly IMongoCollection<TEntity> _dbSet;
private readonly string _collectionName;
private const string _keyField = "_id";
public MongoRepositoryBase(IMongoDbContext mongoDbContext)
{
_dbContext = mongoDbContext;
_collectionName = typeof(TEntity).GetAttributeValue((TableAttribute m) => m.Name)
?? typeof(TEntity).Name;
if (string.IsNullOrWhiteSpace(_collectionName))
throw new ArgumentNullException("Mongo DatabaseName can't be NULL! Please set the attribute Table in your entity class.");
_dbSet = mongoDbContext.GetCollection<TEntity>(_collectionName);
}
#region Create Part
public async Task AddAsync(TEntity entity, IClientSessionHandle session = null)
{
if (session == null)
await _dbSet.InsertOneAsync(entity);
else
await _dbContext.AddCommandAsync(async (session) => await _dbSet.InsertOneAsync(entity));
}
public async Task AddManyAsync(IEnumerable<TEntity> entityList, IClientSessionHandle session = null)
{
if (session == null)
await _dbSet.InsertManyAsync(entityList);
else
await _dbContext.AddCommandAsync(async (session) => await _dbSet.InsertManyAsync(entityList));
}
#endregion
# region Delete Part
public async Task DeleteAsync(string id, IClientSessionHandle session = null)
{
if (session == null)
await _dbSet.DeleteOneAsync(Builders<TEntity>.Filter.Eq(_keyField, new ObjectId(id)));
else
await _dbContext.AddCommandAsync(async (session) => await _dbSet.DeleteOneAsync(Builders<TEntity>.Filter.Eq(_keyField, new ObjectId(id))));
}
public async Task DeleteAsync(Expression<Func<TEntity, bool>> expression, IClientSessionHandle session = null)
{
if (session == null)
await _dbSet.DeleteOneAsync(expression);
else
await _dbContext.AddCommandAsync(async (session) => await _dbSet.DeleteOneAsync(expression));
}
public async Task<DeleteResult> DeleteManyAsync(FilterDefinition<TEntity> filter, IClientSessionHandle session = null)
{
if (session == null)
return await _dbSet.DeleteManyAsync(filter);
await _dbContext.AddCommandAsync(async (session) => await _dbSet.DeleteManyAsync(filter));
return new DeleteResult.Acknowledged(10);
}
public async Task<DeleteResult> DeleteManyAsync(Expression<Func<TEntity, bool>> expression, IClientSessionHandle session = null)
{
if (session == null)
return await _dbSet.DeleteManyAsync(expression);
await _dbContext.AddCommandAsync(async (session) => await _dbSet.DeleteManyAsync(expression));
return new DeleteResult.Acknowledged(10);
}
#endregion
#region Update Part
public async Task UpdateAsync(TEntity entity, IClientSessionHandle session = null)
{
if (session == null)
await _dbSet.ReplaceOneAsync(item => item.Id == entity.Id, entity);
else
await _dbContext.AddCommandAsync(async (session) => await _dbSet.ReplaceOneAsync(item => item.Id == entity.Id, entity));
}
public async Task UpdateAsync(Expression<Func<TEntity, bool>> expression, Expression<Action<TEntity>> entity, IClientSessionHandle session = null)
{
var fieldList = new List<UpdateDefinition<TEntity>>();
if (entity.Body is MemberInitExpression param)
{
foreach (var item in param.Bindings)
{
var propertyName = item.Member.Name;
object propertyValue = null;
if (item is not MemberAssignment memberAssignment) continue;
if (memberAssignment.Expression.NodeType == ExpressionType.Constant)
{
if (memberAssignment.Expression is ConstantExpression constantExpression)
propertyValue = constantExpression.Value;
}
else
{
propertyValue = Expression.Lambda(memberAssignment.Expression, null).Compile().DynamicInvoke();
}
if (propertyName != _keyField)
{
fieldList.Add(Builders<TEntity>.Update.Set(propertyName, propertyValue));
}
}
}
if (session == null)
await _dbSet.UpdateOneAsync(expression, Builders<TEntity>.Update.Combine(fieldList));
else
await _dbContext.AddCommandAsync(async (session) => await _dbSet.UpdateOneAsync(expression, Builders<TEntity>.Update.Combine(fieldList)));
}
public async Task UpdateAsync(FilterDefinition<TEntity> filter, UpdateDefinition<TEntity> update, IClientSessionHandle session = null)
{
if (session == null)
await _dbSet.UpdateOneAsync(filter, update);
else
await _dbContext.AddCommandAsync(async (session) => await _dbSet.UpdateOneAsync(filter, update));
}
public async Task UpdateManyAsync(Expression<Func<TEntity, bool>> expression, UpdateDefinition<TEntity> update, IClientSessionHandle session = null)
{
if (session == null)
await _dbSet.UpdateManyAsync(expression, update);
else
await _dbContext.AddCommandAsync(async (session) => await _dbSet.UpdateManyAsync(expression, update));
}
public async Task<UpdateResult> UpdateManayAsync(Dictionary<string, string> dic, FilterDefinition<TEntity> filter, IClientSessionHandle session = null)
{
var t = new TEntity();
// Fields to be updated
var list = new List<UpdateDefinition<TEntity>>();
foreach (var item in t.GetType().GetProperties())
{
if (!dic.ContainsKey(item.Name)) continue;
var value = dic[item.Name];
list.Add(Builders<TEntity>.Update.Set(item.Name, value));
}
var updatefilter = Builders<TEntity>.Update.Combine(list);
if (session == null)
return await _dbSet.UpdateManyAsync(filter, updatefilter);
await _dbContext.AddCommandAsync(async (session) => await _dbSet.UpdateManyAsync(filter, updatefilter));
return new UpdateResult.Acknowledged(10, 10, null);
}
#endregion
#region Read Part
public async Task<TEntity> GetAsync(Expression<Func<TEntity, bool>> expression, bool readFromPrimary = true)
{
var readPreference = GetReadPreference(readFromPrimary);
var queryData = await _dbSet.WithReadPreference(readPreference)
.Find(expression)
.FirstOrDefaultAsync();
return queryData;
}
public async Task<TEntity> GetAsync(string id, bool readFromPrimary = true)
{
var readPreference = GetReadPreference(readFromPrimary);
var queryData = await _dbSet.WithReadPreference(readPreference).FindAsync(Builders<TEntity>.Filter.Eq(_keyField, new ObjectId(id)));
return queryData.FirstOrDefault();
}
public async Task<IEnumerable<TEntity>> GetAllAsync(bool readFromPrimary = true)
{
var readPreference = GetReadPreference(readFromPrimary);
var queryAllData = await _dbSet.WithReadPreference(readPreference).FindAsync(Builders<TEntity>.Filter.Empty);
return queryAllData.ToList();
}
public async Task<long> CountAsync(Expression<Func<TEntity, bool>> expression, bool readFromPrimary = true)
{
var readPreference = GetReadPreference(readFromPrimary);
return await _dbSet.WithReadPreference(readPreference).CountDocumentsAsync(expression);
}
public async Task<long> CountAsync(FilterDefinition<TEntity> filter, bool readFromPrimary = true)
{
var readPreference = GetReadPreference(readFromPrimary);
return await _dbSet.WithReadPreference(readPreference).CountDocumentsAsync(filter);
}
public async Task<bool> ExistsAsync(Expression<Func<TEntity, bool>> predicate, bool readFromPrimary = true)
{
var readPreference = GetReadPreference(readFromPrimary);
return await Task.FromResult(_dbSet.WithReadPreference(readPreference).AsQueryable().Any(predicate));
}
public async Task<List<TEntity>> FindListAsync(FilterDefinition<TEntity> filter, string[]? field = null, SortDefinition<TEntity>? sort = null, bool readFromPrimary = true)
{
var readPreference = GetReadPreference(readFromPrimary);
if (field == null || field.Length == 0)
{
if (sort == null)
return await _dbSet.WithReadPreference(readPreference).Find(filter).ToListAsync();
return await _dbSet.WithReadPreference(readPreference).Find(filter).Sort(sort).ToListAsync();
}
var fieldList = new List<ProjectionDefinition<TEntity>>();
for (int i = 0; i < field.Length; i++)
{
fieldList.Add(Builders<TEntity>.Projection.Include(field[i].ToString()));
}
var projection = Builders<TEntity>.Projection.Combine(fieldList);
fieldList?.Clear();
if (sort == null)
return await _dbSet.WithReadPreference(readPreference).Find(filter).Project<TEntity>(projection).ToListAsync();
return await _dbSet.WithReadPreference(readPreference).Find(filter).Sort(sort).Project<TEntity>(projection).ToListAsync();
}
public async Task<List<TEntity>> FindListByPageAsync(FilterDefinition<TEntity> filter, int pageIndex, int pageSize, string[]? field = null, SortDefinition<TEntity>? sort = null, bool readFromPrimary = true)
{
var readPreference = GetReadPreference(readFromPrimary);
if (field == null || field.Length == 0)
{
if (sort == null)
return await _dbSet.WithReadPreference(readPreference).Find(filter).Skip((pageIndex - 1) * pageSize).Limit(pageSize).ToListAsync();
return await _dbSet.WithReadPreference(readPreference).Find(filter).Sort(sort).Skip((pageIndex - 1) * pageSize).Limit(pageSize).ToListAsync();
}
var fieldList = new List<ProjectionDefinition<TEntity>>();
for (int i = 0; i < field.Length; i++)
{
fieldList.Add(Builders<TEntity>.Projection.Include(field[i].ToString()));
}
var projection = Builders<TEntity>.Projection.Combine(fieldList);
fieldList?.Clear();
if (sort == null)
return await _dbSet.WithReadPreference(readPreference).Find(filter).Project<TEntity>(projection).Skip((pageIndex - 1) * pageSize).Limit(pageSize).ToListAsync();
return await _dbSet.WithReadPreference(readPreference).Find(filter).Sort(sort).Project<TEntity>(projection).Skip((pageIndex - 1) * pageSize).Limit(pageSize).ToListAsync();
}
#endregion
#region Protected Methods
protected ReadPreference GetReadPreference(bool readFromPrimary)
{
if (readFromPrimary)
return ReadPreference.PrimaryPreferred;
else
return ReadPreference.SecondaryPreferred;
}
#endregion
}
工作单元：UnitOfWork
在实际项目中，在对多个Repository操作之后，我们希望有一个统一的提交操作来实现事务的原子性。因此，我们可以有一个UnitOfWork来作为代理：
public class UnitOfWork : IUnitOfWork
{
private readonly IMongoDbContext _context;
public UnitOfWork(IMongoDbContext context)
{
_context = context;
}
public bool SaveChanges(IClientSessionHandle session)
{
return _context.Commit(session) > 0;
}
public async Task<bool> SaveChangesAsync(IClientSessionHandle session)
{
return await _context.CommitAsync(session) > 0;
}
public IClientSessionHandle BeginTransaction()
{
return _context.StartSession();
}
public async Task<IClientSessionHandle> BeginTransactionAsync()
{
return await _context.StartSessionAsync();
}
public void Dispose()
{
_context.Dispose();
}
}