T2FGame.Persistence.MongoDB 1.0.9

T2FGame.Persistence.MongoDB

T2FGame 框架的 MongoDB 持久化层，提供 Repository 模式实现和工作单元支持。

功能特性

• 通用 Repository：支持 CRUD 和复杂查询
• 聚合根 Repository：集成领域事件分发
• 工作单元：支持 MongoDB 事务（需副本集）
• 连接池管理：高性能连接复用
• 命令日志：可选的调试日志

安装

<PackageReference Include="T2FGame.Persistence.MongoDB" />

快速开始

1. 配置 MongoDB

// 方式1：使用选项配置
services.AddMongoDbPersistence(options =>
{
    options.ConnectionString = "mongodb://localhost:27017";
    options.DatabaseName = "T2FGame";
    options.EnableCommandLogging = true; // 开发时开启
    options.MaxConnectionPoolSize = 100;
});

// 方式2：使用连接字符串
services.AddMongoDbPersistence(
    "mongodb://localhost:27017",
    "T2FGame");

// 方式3：从配置文件读取
services.AddMongoDbPersistence(configuration);

2. appsettings.json 配置

{
  "MongoDB": {
    "ConnectionString": "mongodb://localhost:27017",
    "DatabaseName": "T2FGame",
    "EnableCommandLogging": false,
    "ConnectionTimeoutMs": 10000,
    "ServerSelectionTimeoutMs": 30000,
    "MaxConnectionPoolSize": 100,
    "MinConnectionPoolSize": 10
  }
}

3. 注册 Repository

// 注册通用 Repository
services.AddMongoRepository<Player, long>();
services.AddMongoRepository<Item, string>("game_items"); // 自定义集合名

// 注册聚合根 Repository
services.AddMongoAggregateRepository<Order, Guid>();
services.AddMongoAggregateRepository<UserAggregate, long>("users");

核心组件

MongoDbContext（数据库上下文）

public class MyService
{
    private readonly MongoDbContext _context;

    public MyService(MongoDbContext context)
    {
        _context = context;
    }

    public async Task<bool> CheckConnectionAsync()
    {
        return await _context.PingAsync();
    }

    public IMongoCollection<Player> GetPlayersCollection()
    {
        return _context.GetCollection<Player>();
    }
}

MongoRepository（通用仓储）

public class PlayerService
{
    private readonly IRepository<Player, long> _repository;

    public PlayerService(IRepository<Player, long> repository)
    {
        _repository = repository;
    }

    // 基本 CRUD
    public async Task<Player?> GetByIdAsync(long playerId)
    {
        return await _repository.GetByIdAsync(playerId);
    }

    public async Task<Player> CreateAsync(Player player)
    {
        return await _repository.AddAsync(player);
    }

    public async Task UpdateAsync(Player player)
    {
        await _repository.UpdateAsync(player);
    }

    public async Task DeleteAsync(long playerId)
    {
        await _repository.DeleteByIdAsync(playerId);
    }

    // 查询
    public async Task<IReadOnlyList<Player>> GetByLevelAsync(int minLevel)
    {
        return await _repository.FindAsync(p => p.Level >= minLevel);
    }

    public async Task<Player?> GetByNameAsync(string name)
    {
        return await _repository.FindOneAsync(p => p.Name == name);
    }

    // 统计
    public async Task<long> CountAsync()
    {
        return await _repository.CountAsync();
    }

    public async Task<long> CountOnlineAsync()
    {
        return await _repository.CountAsync(p => p.IsOnline);
    }
}

MongoAggregateRepository（聚合根仓储）

public class OrderService
{
    private readonly IAggregateRepository<Order, Guid> _repository;

    public OrderService(IAggregateRepository<Order, Guid> repository)
    {
        _repository = repository;
    }

    public async Task CreateOrderAsync(Order order)
    {
        // SaveAsync 会自动分发领域事件
        await _repository.SaveAsync(order);
        // order.DomainEvents 中的事件会被发布到 EventBus
    }

    public async Task CompleteOrderAsync(Guid orderId)
    {
        var order = await _repository.GetByIdAsync(orderId);
        if (order == null) throw new NotFoundException();

        order.Complete(); // 这会添加 OrderCompletedEvent
        await _repository.SaveAsync(order); // 事件自动分发
    }
}

MongoUnitOfWork（工作单元）

public class TransferService
{
    private readonly MongoUnitOfWork _unitOfWork;
    private readonly IRepository<Wallet, long> _walletRepository;

    public async Task TransferAsync(long fromId, long toId, decimal amount)
    {
        await _unitOfWork.BeginTransactionAsync();

        try
        {
            var fromWallet = await _walletRepository.GetByIdAsync(fromId);
            var toWallet = await _walletRepository.GetByIdAsync(toId);

            if (fromWallet.Balance < amount)
                throw new InsufficientBalanceException();

            fromWallet.Withdraw(amount);
            toWallet.Deposit(amount);

            await _walletRepository.UpdateAsync(fromWallet);
            await _walletRepository.UpdateAsync(toWallet);

            await _unitOfWork.CommitTransactionAsync();
        }
        catch
        {
            await _unitOfWork.RollbackTransactionAsync();
            throw;
        }
    }
}

自定义 Repository

// 定义接口
public interface IPlayerRepository : IAggregateRepository<Player, long>
{
    Task<Player?> GetByUsernameAsync(string username);
    Task<IReadOnlyList<Player>> GetTopPlayersByScoreAsync(int count);
}

// 实现
public class PlayerRepository : MongoAggregateRepository<Player, long>, IPlayerRepository
{
    public PlayerRepository(
        MongoDbContext context,
        IDomainEventDispatcher? eventDispatcher = null,
        ILogger<PlayerRepository>? logger = null)
        : base(context, eventDispatcher, "players", logger)
    {
    }

    public async Task<Player?> GetByUsernameAsync(string username)
    {
        return await Collection.Find(p => p.Username == username)
            .FirstOrDefaultAsync();
    }

    public async Task<IReadOnlyList<Player>> GetTopPlayersByScoreAsync(int count)
    {
        return await Collection.Find(FilterDefinition<Player>.Empty)
            .SortByDescending(p => p.Score)
            .Limit(count)
            .ToListAsync();
    }
}

// 注册
services.AddCustomRepository<IPlayerRepository, PlayerRepository>();

索引管理

// 在应用启动时创建索引
public class MongoIndexInitializer : IHostedService
{
    private readonly MongoDbContext _context;

    public async Task StartAsync(CancellationToken cancellationToken)
    {
        var players = _context.GetCollection<Player>();

        // 单字段索引
        await players.Indexes.CreateOneAsync(
            new CreateIndexModel<Player>(
                Builders<Player>.IndexKeys.Ascending(p => p.Username),
                new CreateIndexOptions { Unique = true }));

        // 复合索引
        await players.Indexes.CreateOneAsync(
            new CreateIndexModel<Player>(
                Builders<Player>.IndexKeys
                    .Ascending(p => p.Level)
                    .Descending(p => p.Score)));
    }

    public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask;
}

配置选项

public class MongoDbOptions
{
    // 连接设置
    public string ConnectionString { get; set; }
    public string DatabaseName { get; set; }

    // 连接池
    public int MaxConnectionPoolSize { get; set; } = 100;
    public int MinConnectionPoolSize { get; set; } = 10;

    // 超时设置
    public int ConnectionTimeoutMs { get; set; } = 10000;
    public int ServerSelectionTimeoutMs { get; set; } = 30000;

    // 调试
    public bool EnableCommandLogging { get; set; } = false;
}

目录结构

T2FGame.Persistence.MongoDB/
├── Configuration/
│   └── MongoDbOptions.cs            # 配置选项
├── Repositories/
│   ├── MongoRepository.cs           # 通用仓储
│   └── MongoAggregateRepository.cs  # 聚合根仓储
├── Extensions/
│   └── ServiceCollectionExtensions.cs # 服务注册
├── MongoDbContext.cs                # 数据库上下文
├── MongoUnitOfWork.cs               # 工作单元
└── Examples/
    └── ExampleUsage.cs              # 使用示例

注意事项

1. 事务支持：MongoDB 事务需要副本集，单节点不支持
2. ID 映射：Entity 的 Id 会自动映射到 MongoDB 的 _id
3. 并发控制：可使用乐观锁（版本号）或悲观锁
4. 大文档：单文档限制 16MB，大数据考虑 GridFS
5. 连接字符串：生产环境建议使用副本集连接字符串