最近工作中需要用到MongoDB的事務(wù)操作，因此參考了一些資料封裝了一個(gè)小的組件，提供基礎(chǔ)的CRUD Repository基類 和 UnitOfWork工作單元模式。今天，就來(lái)簡(jiǎn)單介紹一下這個(gè)小組件。
關(guān)于MongoDB的事務(wù)
MongoDB在4.2版本開始全面支持了多文檔事務(wù)，至今已過了四年了，雖然我們可能沒有在項(xiàng)目中用MongoDB來(lái)替代傳統(tǒng)關(guān)系型數(shù)據(jù)庫(kù)如MySQL/SQL Server，但是不能否認(rèn)MongoDB已經(jīng)在事務(wù)能力上愈發(fā)成熟了。
在MongoDB中，所謂的事務(wù)主要指的是多個(gè)文檔的事務(wù)，其使用方式和傳統(tǒng)關(guān)系型數(shù)據(jù)庫(kù)差不多。但我們需要注意的是：多文檔事務(wù)只能應(yīng)用在副本集 或 mongos 節(jié)點(diǎn)上。如果你只是一個(gè)單點(diǎn)的mongo實(shí)例，是無(wú)法進(jìn)行多文檔事務(wù)實(shí)踐的。
畫外音：如果你對(duì)MongoDB感興趣，不妨看看我的這個(gè)系列博客：《MongoDB入門到實(shí)踐學(xué)習(xí)之旅》
那么，如何快速進(jìn)行事務(wù)操作呢？
在Mongo Shell中進(jìn)行事務(wù)
下面演示了如何通過Mongo Shell來(lái)進(jìn)行一個(gè)多文檔操作的事務(wù)提交：
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'}});
// 成功提交事務(wù)
session.commitTransaction();
// 失敗事務(wù)回滾
session.abortTransaction();
在.NET應(yīng)用中進(jìn)行事務(wù)
下面展示了在.NET應(yīng)用中通過MongoDB Driver來(lái)進(jìn)行事務(wù)的示例：
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();
}
}
在大部分的實(shí)際應(yīng)用中，我們通常都習(xí)慣使用數(shù)據(jù)倉(cāng)儲(chǔ)（Repository）的模式來(lái)進(jìn)行CRUD，同時(shí)也習(xí)慣用工作單元（UnitOfWork）模式來(lái)進(jìn)行協(xié)調(diào)多個(gè)Repository進(jìn)行事務(wù)提交。那么，如何在自己的項(xiàng)目中實(shí)現(xiàn)這個(gè)呢？
參考了一些資料后，自己實(shí)現(xiàn)了一個(gè)基礎(chǔ)小組件，暫且叫它：EDT.MongoProxy吧，我們來(lái)看看它是如何實(shí)現(xiàn)的。
單例的MongoClient
基于MongoDB的最佳時(shí)間，對(duì)于MongoClient最好設(shè)置為單例注入，因?yàn)樵贛ongoDB.Driver中MongoClient已經(jīng)被設(shè)計(jì)為線程安全可以被多線程共享，這樣可還以避免反復(fù)實(shí)例化MongoClient帶來(lái)的開銷，避免在極端情況下的性能低下。
這里暫且設(shè)計(jì)一個(gè)MongoDbConnection類，用于包裹這個(gè)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;
}
}
其中，這個(gè)MongoDatabaseConfigs類主要是獲取appsettings中的配置，用以生成MongoClient的對(duì)應(yīng)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的設(shè)計(jì)，設(shè)計(jì)一個(gè)MongoDbContext，它從IoC容器中獲取到單例的MongoClient，封裝了事務(wù)的開啟和提交，簡(jiǎn)化了應(yīng)用代碼的編寫。
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);
}
}
數(shù)據(jù)倉(cāng)儲(chǔ)：MongoRepositoryBase
在實(shí)際項(xiàng)目中，我們都希望有一個(gè)基礎(chǔ)的RepositoryBase類，將CRUD的方法都封裝了，我們實(shí)際中就只需要?jiǎng)?chuàng)建一個(gè)對(duì)應(yīng)的Repository集成這個(gè)RepositoryBase就行了，無(wú)需再重復(fù)編寫CRUD的方法。那么，也就有了這個(gè)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
在實(shí)際項(xiàng)目中，在對(duì)多個(gè)Repository操作之后，我們希望有一個(gè)統(tǒng)一的提交操作來(lái)實(shí)現(xiàn)事務(wù)的原子性。因此，我們可以有一個(gè)UnitOfWork來(lái)作為代理：文章來(lái)源：http://www.zghlxwxcb.cn/news/detail-828365.html
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();
}
}
封裝注入：ServiceCollectionExtensions
?文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-828365.html
到了這里，關(guān)于Unity UGUI的PhysicsRayc可變化、可擴(kuò)展aster (物理射線檢測(cè))組件的介紹及使用的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
