大数据

<!-- 项目文件 .csproj --> <PackageReference Include="MongoDB.Driver" Version="2.23.0" />

# 安装主驱动包 Install-Package MongoDB.Driver # 可选：安装异步扩展（推荐） Install-Package MongoDB.Driver.GridFS Install-Package MongoDB.Driver.Core

using MongoDB.Bson; using MongoDB.Bson.Serialization.Attributes; namespace MongoDemo { // 用户实体 public class User { [BsonId] [BsonRepresentation(BsonType.ObjectId)] public string Id { get; set; } [BsonElement("name")] public string Name { get; set; } [BsonElement("email")] public string Email { get; set; } [BsonElement("age")] public int Age { get; set; } [BsonElement("address")] public Address Address { get; set; } [BsonElement("tags")] public List<string> Tags { get; set; } = new(); [BsonElement("created_at")] [BsonDateTimeOptions(Kind = DateTimeKind.Local)] public DateTime CreatedAt { get; set; } } // 嵌套地址类 public class Address { [BsonElement("city")] public string City { get; set; } [BsonElement("street")] public string Street { get; set; } [BsonElement("postal_code")] public string PostalCode { get; set; } } // 产品实体 public class Product { [BsonId] [BsonRepresentation(BsonType.ObjectId)] public string Id { get; set; } public string Name { get; set; } public decimal Price { get; set; } public int Stock { get; set; } public List<string> Categories { get; set; } public Dictionary<string, object> Attributes { get; set; } } }

using MongoDB.Driver; using System; namespace MongoDemo { public class MongoService { private readonly IMongoDatabase _database; // 方式1：基础连接 public MongoService() { var client = new MongoClient("mongodb://localhost:27017"); _database = client.GetDatabase("testdb"); } // 方式2：带认证的连接 public MongoService(string connectionString) { var client = new MongoClient(connectionString); _database = client.GetDatabase("testdb"); } // 方式3：复杂配置连接 public MongoService() { var settings = new MongoClientSettings { Servers = new[] { new MongoServerAddress("localhost", 27017) }, Credential = MongoCredential.CreateCredential( "admin", "username", "password" ), // 连接池配置 MaxConnectionPoolSize = 100, MinConnectionPoolSize = 10, // 读写配置 ReadPreference = ReadPreference.SecondaryPreferred, WriteConcern = WriteConcern.WMajority }; var client = new MongoClient(settings); _database = client.GetDatabase("testdb"); } // 获取集合 public IMongoCollection<User> Users => _database.GetCollection<User>("users"); public IMongoCollection<Product> Products => _database.GetCollection<Product>("products"); } }

public class UserRepository { private readonly IMongoCollection<User> _users; public UserRepository(IMongoDatabase database) { _users = database.GetCollection<User>("users"); } // CREATE - 插入文档 public async Task<User> CreateUserAsync(User user) { user.Id = ObjectId.GenerateNewId().ToString(); user.CreatedAt = DateTime.Now; await _users.InsertOneAsync(user); return user; } public async Task CreateMultipleUsersAsync(List<User> users) { foreach (var user in users) { user.Id = ObjectId.GenerateNewId().ToString(); user.CreatedAt = DateTime.Now; } await _users.InsertManyAsync(users); } // READ - 查询文档 public async Task<User> GetUserByIdAsync(string id) { var filter = Builders<User>.Filter.Eq(u => u.Id, id); return await _users.Find(filter).FirstOrDefaultAsync(); } public async Task<List<User>> GetUsersByAgeAsync(int minAge, int maxAge) { var filter = Builders<User>.Filter.And( Builders<User>.Filter.Gte(u => u.Age, minAge), Builders<User>.Filter.Lte(u => u.Age, maxAge) ); return await _users.Find(filter) .SortBy(u => u.Name) .Limit(100) .ToListAsync(); } public async Task<User> FindUserByEmailAsync(string email) { var filter = Builders<User>.Filter.Eq(u => u.Email, email); return await _users.Find(filter).FirstOrDefaultAsync(); } // UPDATE - 更新文档 public async Task<bool> UpdateUserEmailAsync(string userId, string newEmail) { var filter = Builders<User>.Filter.Eq(u => u.Id, userId); var update = Builders<User>.Update .Set(u => u.Email, newEmail) .Set(u => u.UpdatedAt, DateTime.Now); var result = await _users.UpdateOneAsync(filter, update); return result.ModifiedCount > 0; } public async Task<long> IncrementUserAgeAsync(string userId) { var filter = Builders<User>.Filter.Eq(u => u.Id, userId); var update = Builders<User>.Update.Inc(u => u.Age, 1); var result = await _users.UpdateOneAsync(filter, update); return result.ModifiedCount; } public async Task<bool> AddTagToUserAsync(string userId, string tag) { var filter = Builders<User>.Filter.Eq(u => u.Id, userId); var update = Builders<User>.Update.AddToSet(u => u.Tags, tag); var result = await _users.UpdateOneAsync(filter, update); return result.ModifiedCount > 0; } // DELETE - 删除文档 public async Task<bool> DeleteUserAsync(string userId) { var filter = Builders<User>.Filter.Eq(u => u.Id, userId); var result = await _users.DeleteOneAsync(filter); return result.DeletedCount > 0; } public async Task<long> DeleteUsersByAgeAsync(int age) { var filter = Builders<User>.Filter.Eq(u => u.Age, age); var result = await _users.DeleteManyAsync(filter); return result.DeletedCount; } }

public class LinqRepository { private readonly IMongoCollection<User> _users; public LinqRepository(IMongoDatabase database) { _users = database.GetCollection<User>("users"); } // LINQ 查询示例 public async Task<List<User>> GetUsersByLinqAsync() { // 基本查询 var query = from user in _users.AsQueryable() where user.Age > 18 && user.Email.Contains("@gmail.com") orderby user.CreatedAt descending select user; return await query.ToListAsync(); } public async Task<List<User>> GetUsersInCityAsync(string city) { return await _users.AsQueryable() .Where(u => u.Address.City == city) .Where(u => u.Tags.Contains("vip")) .OrderBy(u => u.Name) .Take(50) .ToListAsync(); } public async Task<User> FindUserByEmailLinqAsync(string email) { return await _users.AsQueryable() .FirstOrDefaultAsync(u => u.Email == email); } // 投影查询（只获取部分字段） public async Task<List<object>> GetUserNamesAndEmailsAsync() { return await _users.AsQueryable() .Where(u => u.Age > 20) .Select(u => new { u.Name, u.Email, u.Age }) .ToListAsync(); } }

public class AdvancedOperations { private readonly IMongoCollection<User> _users; public async Task<List<User>> ComplexQueryAsync() { // 构建复杂过滤条件 var filter = Builders<User>.Filter.And( Builders<User>.Filter.Gte(u => u.Age, 18), Builders<User>.Filter.Lte(u => u.Age, 65), Builders<User>.Filter.ElemMatch(u => u.Tags, Builders<string>.Filter.In("tag", new[] { "developer", "designer" })), Builders<User>.Filter.Or( Builders<User>.Filter.Eq(u => u.Address.City, "北京"), Builders<User>.Filter.Eq(u => u.Address.City, "上海") ) ); // 构建排序和分页 var sort = Builders<User>.Sort .Ascending(u => u.Age) .Descending(u => u.CreatedAt); return await _users.Find(filter) .Sort(sort) .Skip(0) .Limit(20) .ToListAsync(); } // 聚合管道示例 public async Task<List<BsonDocument>> AggregateUsersAsync() { var pipeline = new[] { new BsonDocument("$match", new BsonDocument("age", new BsonDocument("$gte", 18))), new BsonDocument("$group", new BsonDocument { { "_id", "$address.city" }, { "count", new BsonDocument("$sum", 1) }, { "averageAge", new BsonDocument("$avg", "$age") }, { "maxAge", new BsonDocument("$max", "$age") } }), new BsonDocument("$sort", new BsonDocument("count", -1)), new BsonDocument("$limit", 10) }; return await _users.Aggregate<BsonDocument>(pipeline).ToListAsync(); } // 使用强类型聚合 public class CityStats { public string City { get; set; } public int Count { get; set; } public double AverageAge { get; set; } } public async Task<List<CityStats>> AggregateUsersTypedAsync() { var pipeline = _users.Aggregate() .Match(u => u.Age >= 18) .Group(u => u.Address.City, g => new CityStats { City = g.Key, Count = g.Count(), AverageAge = g.Average(u => u.Age) }) .SortByDescending(s => s.Count) .Limit(10); return await pipeline.ToListAsync(); } }

// Program.cs using MongoDB.Driver; var builder = WebApplication.CreateBuilder(args); // 配置 MongoDB builder.Services.Configure<MongoDBSettings>( builder.Configuration.GetSection("MongoDB")); // 注册 MongoDB 服务 builder.Services.AddSingleton<IMongoClient>(serviceProvider => { var settings = serviceProvider.GetRequiredService<IOptions<MongoDBSettings>>().Value; return new MongoClient(settings.ConnectionString); }); builder.Services.AddScoped<IMongoDatabase>(serviceProvider => { var client = serviceProvider.GetRequiredService<IMongoClient>(); var settings = serviceProvider.GetRequiredService<IOptions<MongoDBSettings>>().Value; return client.GetDatabase(settings.DatabaseName); }); // 注册仓储 builder.Services.AddScoped<IUserRepository, UserRepository>(); builder.Services.AddScoped<IProductRepository, ProductRepository>(); var app = builder.Build();

{ "MongoDB": { "ConnectionString": "mongodb://localhost:27017", "DatabaseName": "myappdb", "UsersCollection": "users", "ProductsCollection": "products" }, "Logging": { "LogLevel": { "Default": "Information" } } }

public class MongoDBSettings { public string ConnectionString { get; set; } public string DatabaseName { get; set; } public string UsersCollection { get; set; } public string ProductsCollection { get; set; } }

public interface IUserRepository { Task<User> GetByIdAsync(string id); Task<IEnumerable<User>> GetAllAsync(); Task<User> CreateAsync(User user); Task UpdateAsync(User user); Task DeleteAsync(string id); } public class UserRepository : IUserRepository { private readonly IMongoCollection<User> _users; public UserRepository(IMongoDatabase database, IOptions<MongoDBSettings> settings) { _users = database.GetCollection<User>( settings.Value.UsersCollection); // 创建索引 CreateIndexes(); } private void CreateIndexes() { // 创建唯一索引 var emailIndex = Builders<User>.IndexKeys.Ascending(u => u.Email); var emailIndexOptions = new CreateIndexOptions { Unique = true }; _users.Indexes.CreateOne( new CreateIndexModel<User>(emailIndex, emailIndexOptions)); // 创建复合索引 var compoundIndex = Builders<User>.IndexKeys .Ascending(u => u.Address.City) .Ascending(u => u.Age); _users.Indexes.CreateOne( new CreateIndexModel<User>(compoundIndex)); } public async Task<User> GetByIdAsync(string id) { return await _users.Find(u => u.Id == id).FirstOrDefaultAsync(); } public async Task<IEnumerable<User>> GetAllAsync() { return await _users.Find(_ => true).ToListAsync(); } public async Task<User> CreateAsync(User user) { await _users.InsertOneAsync(user); return user; } public async Task UpdateAsync(User user) { await _users.ReplaceOneAsync(u => u.Id == user.Id, user); } public async Task DeleteAsync(string id) { await _users.DeleteOneAsync(u => u.Id == id); } }

[ApiController] [Route("api/[controller]")] public class UsersController : ControllerBase { private readonly IUserRepository _userRepository; public UsersController(IUserRepository userRepository) { _userRepository = userRepository; } [HttpGet] public async Task<IActionResult> GetAll() { var users = await _userRepository.GetAllAsync(); return Ok(users); } [HttpGet("{id}")] public async Task<IActionResult> GetById(string id) { var user = await _userRepository.GetByIdAsync(id); if (user == null) return NotFound(); return Ok(user); } [HttpPost] public async Task<IActionResult> Create([FromBody] CreateUserDto dto) { var user = new User { Name = dto.Name, Email = dto.Email, Age = dto.Age, Address = dto.Address, CreatedAt = DateTime.Now }; var createdUser = await _userRepository.CreateAsync(user); return CreatedAtAction(nameof(GetById), new { id = createdUser.Id }, createdUser); } [HttpPut("{id}")] public async Task<IActionResult> Update(string id, [FromBody] UpdateUserDto dto) { var existingUser = await _userRepository.GetByIdAsync(id); if (existingUser == null) return NotFound(); existingUser.Name = dto.Name; existingUser.Email = dto.Email; existingUser.Age = dto.Age; await _userRepository.UpdateAsync(existingUser); return NoContent(); } [HttpDelete("{id}")] public async Task<IActionResult> Delete(string id) { await _userRepository.DeleteAsync(id); return NoContent(); } // 高级查询端点 [HttpGet("search")] public async Task<IActionResult> Search( [FromQuery] string city, [FromQuery] int? minAge, [FromQuery] int? maxAge, [FromQuery] string tag) { var builder = Builders<User>.Filter; var filter = builder.Empty; if (!string.IsNullOrEmpty(city)) filter &= builder.Eq(u => u.Address.City, city); if (minAge.HasValue) filter &= builder.Gte(u => u.Age, minAge.Value); if (maxAge.HasValue) filter &= builder.Lte(u => u.Age, maxAge.Value); if (!string.IsNullOrEmpty(tag)) filter &= builder.AnyEq(u => u.Tags, tag); // 实际实现中，这里需要访问具体的MongoCollection // 简化示例... return Ok(); } } // DTO 类 public class CreateUserDto { public string Name { get; set; } public string Email { get; set; } public int Age { get; set; } public Address Address { get; set; } } public class UpdateUserDto { public string Name { get; set; } public string Email { get; set; } public int Age { get; set; } }

public async Task<bool> TransferPointsAsync( string fromUserId, string toUserId, int points) { var client = _database.Client; using (var session = await client.StartSessionAsync()) { session.StartTransaction(); try { // 扣减源用户积分 var fromFilter = Builders<User>.Filter.Eq(u => u.Id, fromUserId); var fromUpdate = Builders<User>.Update.Inc(u => u.Points, -points); await _users.UpdateOneAsync(session, fromFilter, fromUpdate); // 增加目标用户积分 var toFilter = Builders<User>.Filter.Eq(u => u.Id, toUserId); var toUpdate = Builders<User>.Update.Inc(u => u.Points, points); await _users.UpdateOneAsync(session, toFilter, toUpdate); // 提交事务 await session.CommitTransactionAsync(); return true; } catch (Exception ex) { await session.AbortTransactionAsync(); Console.WriteLine($"Transaction failed: {ex.Message}"); return false; } } }

public async Task BulkInsertUsersAsync(List<User> users) { // 使用 BulkWrite 提高性能 var writes = new List<WriteModel<User>>(); foreach (var user in users) { var filter = Builders<User>.Filter.Eq(u => u.Email, user.Email); var update = Builders<User>.Update .SetOnInsert(u => u.Id, ObjectId.GenerateNewId().ToString()) .Set(u => u.Name, user.Name) .Set(u => u.Age, user.Age) .Set(u => u.CreatedAt, DateTime.Now); writes.Add(new UpdateOneModel<User>(filter, update) { IsUpsert = true }); } var options = new BulkWriteOptions { IsOrdered = false }; await _users.BulkWriteAsync(writes, options); }

public class MongoServiceWithLogging { private readonly IMongoCollection<User> _users; private readonly ILogger<MongoServiceWithLogging> _logger; public MongoServiceWithLogging( IMongoDatabase database, ILogger<MongoServiceWithLogging> logger) { _users = database.GetCollection<User>("users"); _logger = logger; // 配置命令监视（开发环境） var client = database.Client; client.Settings.ClusterConfigurator = cb => { cb.Subscribe<CommandStartedEvent>(e => { _logger.LogDebug( "MongoDB Command: {CommandName} - {CommandJson}", e.CommandName, e.Command.ToJson()); }); cb.Subscribe<CommandSucceededEvent>(e => { _logger.LogDebug( "MongoDB Command succeeded in {Duration}ms", e.Duration.TotalMilliseconds); }); }; } }

public class IndexManager { private readonly IMongoCollection<User> _users; public async Task EnsureIndexesAsync() { // 1. 唯一索引 var emailIndex = Builders<User>.IndexKeys.Ascending(u => u.Email); await _users.Indexes.CreateOneAsync( new CreateIndexModel<User>(emailIndex, new CreateIndexOptions { Unique = true })); // 2. 复合索引 var compoundIndex = Builders<User>.IndexKeys .Ascending(u => u.Address.City) .Ascending(u => u.Age) .Descending(u => u.CreatedAt); await _users.Indexes.CreateOneAsync( new CreateIndexModel<User>(compoundIndex)); // 3. TTL索引（自动过期） var ttlIndex = Builders<User>.IndexKeys.Ascending(u => u.CreatedAt); await _users.Indexes.CreateOneAsync( new CreateIndexModel<User>(ttlIndex, new CreateIndexOptions { ExpireAfter = TimeSpan.FromDays(30) })); // 4. 文本索引（全文搜索） var textIndex = Builders<User>.IndexKeys.Text(u => u.Name); await _users.Indexes.CreateOneAsync( new CreateIndexModel<User>(textIndex)); } public async Task<List<string>> GetIndexesAsync() { using (var cursor = await _users.Indexes.ListAsync()) { var indexes = await cursor.ToListAsync(); return indexes.Select(i => i["name"].AsString).ToList(); } } }

var settings = new MongoClientSettings { Servers = new[] { new MongoServerAddress("localhost", 27017) }, // 连接池配置 MaxConnectionPoolSize = 100, MinConnectionPoolSize = 10, MaxConnectionIdleTime = TimeSpan.FromMinutes(5), MaxConnectionLifeTime = TimeSpan.FromMinutes(30), // 等待连接超时 WaitQueueTimeout = TimeSpan.FromSeconds(30), // 连接超时 ConnectTimeout = TimeSpan.FromSeconds(10), // 套接字超时 SocketTimeout = TimeSpan.FromSeconds(30) };

// 优化前：查询所有字段 var users = await _users.Find(_ => true).ToListAsync(); // 优化后：只查询需要的字段 var projection = Builders<User>.Projection .Include(u => u.Name) .Include(u => u.Email) .Exclude(u => u.Id); var users = await _users.Find(_ => true) .Project<User>(projection) .ToListAsync(); // 使用游标处理大量数据 using (var cursor = await _users.Find(_ => true) .BatchSize(1000) // 每批获取1000条 .ToCursorAsync()) { while (await cursor.MoveNextAsync()) { foreach (var user in cursor.Current) { // 处理数据 } } }

MongoDB.Driver                 # 主驱动包
MongoDB.Driver.GridFS          # GridFS文件存储
MongoDB.Driver.Core            # 核心驱动
MongoDB.Bson                   # BSON序列化

MongoDemo/
├── Models/                    # 实体类
│   ├── User.cs
│   ├── Product.cs
│   └── Address.cs
├── Repositories/              # 数据访问层
│   ├── IUserRepository.cs
│   └── UserRepository.cs
├── Services/                  # 业务逻辑层
│   └── UserService.cs
├── Controllers/               # Web API
│   └── UsersController.cs
├── DTOs/                      # 数据传输对象
│   ├── CreateUserDto.cs
│   └── UpdateUserDto.cs
└── Program.cs                 # 配置和启动

// 本地无认证 "mongodb://localhost:27017" // 带认证 "mongodb://username:password@localhost:27017/dbname" // 副本集 "mongodb://host1:27017,host2:27017,host3:27017/dbname?replicaSet=rs0" // Atlas云服务 "mongodb+srv://username:password@cluster0.mongodb.net/dbname"

// 全局序列化配置 BsonClassMap.RegisterClassMap<User>(cm => { cm.AutoMap(); cm.MapMember(u => u.CreatedAt) .SetSerializer(new DateTimeSerializer(DateTimeKind.Local)); cm.UnmapMember(u => u.InternalField); // 不序列化某些字段 }); // 自定义序列化器 public class CustomDateTimeSerializer : DateTimeSerializer { public override DateTime Deserialize(...) { // 自定义逻辑 } }

try { var user = await _users.Find(u => u.Email == email).FirstAsync(); return user; } catch (MongoException ex) when (ex is MongoCommandException || ex is MongoWriteException) { // 处理重复键等错误 _logger.LogError(ex, "MongoDB操作失败"); throw new ApplicationException("数据库操作失败", ex); } catch (TimeoutException) { // 处理超时 throw new ApplicationException("数据库连接超时"); }