Name: EasilyNET.Mongo.AspNetCore
Author: joesdu

EasilyNET.Mongo.AspNetCore — nugetz

// ❌ 旧方式（不再支持）
public class MyService(IMongoClient client, IMongoDatabase database) { }

// ✅ 新方式：通过 DbContext 获取
public class MyService(MyDbContext db)
{
    // 访问 IMongoClient
    var client = db.Client;

    // 访问 IMongoDatabase
    var database = db.Database;

    // 直接使用集合
    var orders = db.Orders;
}

// ❌ 旧方式（不再支持）
builder.Services.AddMongoContext<MyDbContext>(config, c =>
{
    c.DefaultConventionRegistry = true;
    c.ObjectIdToStringTypes = [typeof(SomeEntity)];
    c.ConventionRegistry = new() { ... };
});

// ✅ 新方式：Convention 独立配置，在 AddMongoContext 之前调用
builder.Services.ConfigureMongoConventions(c =>
{
    c.ObjectIdToStringTypes = [typeof(SomeEntity)];
    c.AddConvention("myConvention", new ConventionPack { ... });
});
builder.Services.AddMongoContext<MyDbContext>(config, c =>
{
    c.DatabaseName = "mydb";
});

public class MyDbContext : MongoContext
{
    public IMongoCollection<Order> Orders { get; set; } = default!;
    public IMongoCollection<User> Users { get; set; } = default!;
}

{
  "ConnectionStrings": {
    "Mongo": "mongodb://localhost:27017/mydb"
  }
}

var builder = WebApplication.CreateBuilder(args);

// 可选：自定义全局 Convention（必须在 AddMongoContext 之前调用，最多一次）
// 调用后仅使用用户自定义的约定，本库内置默认约定不会被应用
// 若不调用，首次 AddMongoContext 时自动使用默认配置
builder.Services.ConfigureMongoConventions(c =>
{
    // 特定类型的 ObjectId 存为 string（在使用 $unwind 时有时需要此配置）
    c.ObjectIdToStringTypes = [typeof(SomeEntity)];

    // 添加自定义约定包（支持链式调用）
    c.AddConvention("default", new ConventionPack
    {
        new CamelCaseElementNameConvention(),
        new IgnoreExtraElementsConvention(true),
        new NamedIdMemberConvention("Id", "ID"),
        new EnumRepresentationConvention(BsonType.String)
    })
    .AddConvention("ignoreDefault", new ConventionPack
    {
        new IgnoreIfDefaultConvention(true)
    });
});

builder.Services.AddMongoContext<MyDbContext>(builder.Configuration, c =>
{
    // 数据库名称（可选，覆盖连接字符串中的库名）
    c.DatabaseName = "mydb";

    // 高级驱动配置（可选，如对接 APM 探针）
    c.ClientSettings = cs =>
    {
        cs.ClusterConfigurator = cb => cb.Subscribe(new DiagnosticsActivityEventSubscriber());
    };
});

builder.Services.AddMongoContext<MyDbContext>("mongodb://localhost:27017/mydb", c =>
{
    c.DatabaseName = "mydb";
});

builder.Services.AddMongoContext<MyDbContext>(
    new MongoClientSettings
    {
        Servers = [new MongoServerAddress("127.0.0.1", 27017)],
        Credential = MongoCredential.CreateCredential("admin", "username", "password")
    },
    c =>
    {
        c.DatabaseName = "mydb";
    });

builder.Services.AddMongoContext<MyDbContext>(builder.Configuration, c =>
{
    // 启用弹性默认值（与驱动内置自动恢复机制配合）
    c.Resilience.Enable = true;

    // 以下均为可选调整，括号内为默认值
    c.Resilience.ServerSelectionTimeout = TimeSpan.FromSeconds(10); // 服务器选择超时
    c.Resilience.ConnectTimeout = TimeSpan.FromSeconds(10);         // TCP 连接建立超时
    c.Resilience.SocketTimeout = TimeSpan.FromSeconds(60);          // Socket 读写超时
    c.Resilience.WaitQueueTimeout = TimeSpan.FromMinutes(1);        // 连接池等待超时
    c.Resilience.HeartbeatInterval = TimeSpan.FromSeconds(10);      // 心跳检测间隔
    c.Resilience.MaxConnectionPoolSize = 100;                       // 最大连接数
    c.Resilience.MinConnectionPoolSize = null;                      // 最小连接数（null = 使用驱动默认值）
    c.Resilience.RetryReads = true;                                 // 自动重试读操作
    c.Resilience.RetryWrites = true;                                // 自动重试写操作
});

场景	推荐配置
同区域低延迟	`ConnectTimeout=5s`, `ServerSelectionTimeout=5s`
跨区域高延迟	`ConnectTimeout=20s`, `ServerSelectionTimeout=30s`
高并发大流量	`MaxConnectionPoolSize=200+`, `WaitQueueTimeout=30s`
代理 / 单节点	连接串加 `directConnection=true` 或 `loadBalanced=true`
Atlas / 云托管	保持默认值，确保 `RetryReads=true`, `RetryWrites=true`

功能	说明	示例
驼峰字段名	C# `PascalCase` → MongoDB `camelCase`	`PageSize` → `pageSize`
`_id` 映射	自动将 `_id` 与实体中的 `Id` / `ID` 属性互相映射	`_id` ↔ `Id`
枚举存字符串	枚举值以字符串形式存储，便于阅读	`Gender.Male` → `"Male"`
忽略未知字段	反序列化时忽略数据库中存在但代码中未定义的字段	向前兼容
DateTime 本地化	`DateTime` 反序列化后自动设为 `DateTimeKind.Local`	时区一致
Decimal128	`decimal` 类型自动映射为 MongoDB `Decimal128`，避免精度丢失	金额字段

// 字符串格式（默认 "yyyy-MM-dd" 和 "HH:mm:ss.ffffff"，便于阅读和查询）
builder.Services.RegisterSerializer(new DateOnlySerializerAsString());
builder.Services.RegisterSerializer(new TimeOnlySerializerAsString());

// 自定义格式
builder.Services.RegisterSerializer(new DateOnlySerializerAsString("yyyy/MM/dd"));

// Ticks 格式（long 类型，节省空间，适合高频时间字段）
builder.Services.RegisterSerializer(new DateOnlySerializerAsTicks());
builder.Services.RegisterSerializer(new TimeOnlySerializerAsTicks());

// object、dynamic、匿名类型支持
builder.Services.RegisterDynamicSerializer();

// System.Text.Json 的 JsonNode / JsonObject 类型
builder.Services.RegisterSerializer(new JsonNodeSerializer());
builder.Services.RegisterSerializer(new JsonObjectSerializer());

// 支持 Dictionary<TEnum, TValue> / IDictionary<TEnum, TValue> / IReadOnlyDictionary<TEnum, TValue>
builder.Services.RegisterGlobalEnumKeyDictionarySerializer();

builder.Services.RegisterSerializer(new DoubleSerializer(BsonType.Double));
builder.Services.RegisterSerializer(new GuidSerializer(GuidRepresentation.Standard));

var app = builder.Build();

// 自动扫描所有带索引特性的实体，后台创建/更新索引（不阻塞应用启动）
app.UseCreateMongoIndexes<MyDbContext>();

builder.Services.AddMongoContext<MyDbContext>(builder.Configuration, c =>
{
    // 启用自动删除未管理的索引（谨慎！会删除 DBA 手动创建的索引）
    c.DropUnmanagedIndexes = true;

    // 保护特定前缀的索引不被删除（即使 DropUnmanagedIndexes = true）
    c.ProtectedIndexPrefixes.Add("dba_");      // 保护 DBA 手动创建的索引
    c.ProtectedIndexPrefixes.Add("analytics_"); // 保护分析用索引
});

[TimeSeriesCollection(
    collectionName: "sensor_readings",
    timeField: "timestamp",      // 时间字段（DateTime 类型）
    metaField: "deviceId",       // 分组字段（传感器ID、设备ID 等）
    granularity: TimeSeriesGranularity.Seconds)]
public class SensorReading
{
    public string Id { get; set; }
    public DateTime Timestamp { get; set; }
    public string DeviceId { get; set; }
    public double Temperature { get; set; }
}

// 若集合不存在则自动创建，已存在则跳过
app.UseCreateMongoTimeSeriesCollection<MyDbContext>();

// 保存最近 100MB 的操作日志
[CappedCollection(collectionName: "operation_logs", maxSize: 100 * 1024 * 1024)]
public class OperationLog
{
    public string Id { get; set; }
    public DateTime CreatedAt { get; set; }
    public string Action { get; set; }
}

// 同时限制大小和条数（二者均满足才覆盖旧数据）
[CappedCollection("audit_logs", maxSize: 50 * 1024 * 1024, MaxDocuments = 50000)]
public class AuditLog
{
    // ...
}

app.UseCreateMongoCappedCollections<MyDbContext>();

[MongoSearchIndex(Name = "product_search")]
[MongoSearchIndex(Name = "product_vector", Type = ESearchIndexType.VectorSearch)]
public class Product
{
    public string Id { get; set; }

    // 中文全文搜索 + 自动补全
    [SearchField(ESearchFieldType.String, IndexName = "product_search",
        AnalyzerName = "lucene.chinese")]
    [SearchField(ESearchFieldType.Autocomplete, IndexName = "product_search",
        AnalyzerName = "lucene.chinese")]
    public string Name { get; set; }

    // 1536 维向量（对应 OpenAI text-embedding-ada-002）
    [VectorField(Dimensions = 1536, Similarity = EVectorSimilarity.Cosine,
        IndexName = "product_vector")]
    public float[] Embedding { get; set; }

    // 向量搜索的预过滤字段（只在指定分类中搜索）
    [VectorFilterField(IndexName = "product_vector")]
    public string Category { get; set; }
}

// 该实体不需要在 DbContext 中声明，框架通过程序集扫描自动发现
[MongoSearchIndex(Name = "log_search", CollectionName = "application_logs")]
public class ApplicationLog
{
    [SearchField(ESearchFieldType.String, AnalyzerName = "lucene.standard")]
    public string Message { get; set; }

    [SearchField(ESearchFieldType.Date)]
    public DateTime Timestamp { get; set; }
}

// 方式 1（推荐）：在服务注册阶段添加后台服务，自动创建索引
builder.Services.AddMongoSearchIndexCreation<MyDbContext>();

// 方式 2：在中间件管道中调用（异步后台创建，不阻塞应用启动）
app.UseCreateMongoSearchIndexes<MyDbContext>();

// 先用 AI 模型生成查询向量，再进行语义搜索
var queryVector = await embeddingService.GetEmbeddingAsync("蓝牙耳机 降噪");

var pipeline = new BsonDocument[]
{
    new("$vectorSearch", new BsonDocument
    {
        { "index", "product_vector" },
        { "path", "embedding" },
        { "queryVector", new BsonArray(queryVector.Select(f => (BsonValue)f)) },
        { "numCandidates", 150 },
        { "limit", 10 },
        { "filter", new BsonDocument("category", "电子产品") }  // 预过滤
    })
};

var results = await db.Products.Aggregate<BsonDocument>(pipeline).ToListAsync();

using EasilyNET.Mongo.AspNetCore.ChangeStreams;
using EasilyNET.Mongo.AspNetCore.Options;

public class OrderChangeStreamHandler : MongoChangeStreamHandler<Order>
{
    private readonly IServiceScopeFactory _scopeFactory;

    public OrderChangeStreamHandler(
        MyDbContext db,
        ILogger<OrderChangeStreamHandler> logger,
        IServiceScopeFactory scopeFactory)
        : base(db.Database, collectionName: "orders", logger, new ChangeStreamHandlerOptions
        {
            // 持久化恢复令牌：应用重启后从上次位置继续，不丢失事件
            PersistResumeToken = true,
            ResumeTokenCollectionName = "_changeStreamResumeTokens", // 令牌存储集合

            // 断线重连：指数退避策略
            MaxRetryAttempts = 5,                         // 最大重试次数（0 = 无限）
            RetryDelay = TimeSpan.FromSeconds(2),         // 初始间隔（2→4→8→16→32s）
            MaxRetryDelay = TimeSpan.FromSeconds(60),     // 最大间隔

            // FullDocument 策略（更新事件是否拉取完整文档）
            FullDocument = ChangeStreamFullDocumentOption.UpdateLookup,
        })
    {
        _scopeFactory = scopeFactory;
    }

    // 只监听插入和更新（不设置则监听全部操作类型）
    protected override ChangeStreamOperationType[]? WatchOperations =>
    [
        ChangeStreamOperationType.Insert,
        ChangeStreamOperationType.Update,
        ChangeStreamOperationType.Replace
    ];

    protected override async Task HandleChangeAsync(
        ChangeStreamDocument<Order> change,
        CancellationToken cancellationToken)
    {
        // 处理器本身是 Singleton，Scoped 服务需通过 Scope 获取
        using var scope = _scopeFactory.CreateScope();
        var notifier = scope.ServiceProvider.GetRequiredService<INotificationService>();

        switch (change.OperationType)
        {
            case ChangeStreamOperationType.Insert:
                await notifier.SendNewOrderAlertAsync(change.FullDocument, cancellationToken);
                break;

            case ChangeStreamOperationType.Update when change.FullDocument?.Status == "shipped":
                await notifier.SendShippedNotificationAsync(change.FullDocument, cancellationToken);
                break;
        }
    }
}

// 注册为后台服务，应用启动时自动开始监听
builder.Services.AddMongoChangeStreamHandler<OrderChangeStreamHandler>();

属性	默认值	说明
`MaxRetryAttempts`	`5`	断线后最大重试次数，`0` 表示无限重试
`RetryDelay`	`2s`	首次重试间隔，后续每次翻倍
`MaxRetryDelay`	`60s`	重试间隔上限
`PersistResumeToken`	`false`	是否将恢复令牌持久化到 MongoDB
`ResumeTokenCollectionName`	`"_changeStreamResumeTokens"`	存储恢复令牌的集合名
`FullDocument`	`UpdateLookup`	更新事件是否返回完整文档

开启 PersistResumeToken = true 后：

事件 1 → HandleChangeAsync() → 保存 Token-A
事件 2 → HandleChangeAsync() → 保存 Token-B
[应用重启]
从 Token-B 恢复 → 继续处理事件 3、4、5 ...（无遗漏，无重复）

// 使用默认数据库（集合：fs.files, fs.chunks）
builder.Services.AddGridFSBucket<MyDbContext>();

// 自定义桶名和块大小
builder.Services.AddGridFSBucket<MyDbContext>(opt =>
{
    opt.BucketName = "uploads";      // 集合前缀：uploads.files, uploads.chunks
    opt.ChunkSizeBytes = 512 * 1024;  // 每块 512KB（默认 255KB）
});

// 使用独立的数据库（文件库与业务库分离），通过键控服务注入
builder.Services.AddGridFSBucket<MyDbContext>(
    serviceKey: "media",              // DI 键名，注入时使用 [FromKeyedServices("media")]
    databaseName: "file-storage-db",  // 独立数据库名
    opt =>
    {
        opt.BucketName = "media";
    });

// 注册多个桶
builder.Services.AddGridFSBucket<MyDbContext>("media", "media-db");
builder.Services.AddGridFSBucket<MyDbContext>("documents", "docs-db");

// 在服务中注入
public class MediaService([FromKeyedServices("media")] IGridFSBucket mediaBucket)
{
    // mediaBucket 操作 media-db 数据库中的 fs.files / fs.chunks
}

public class FileStorageService(IGridFSBucket gridFs)
{
    // 上传文件
    public async Task<string> UploadAsync(string fileName, Stream content, CancellationToken ct = default)
    {
        var options = new GridFSUploadOptions
        {
            Metadata = new BsonDocument
            {
                { "contentType", "image/jpeg" },
                { "uploadedBy", "user_001" }
            }
        };
        var fileId = await gridFs.UploadFromStreamAsync(fileName, content, options, ct);
        return fileId.ToString();
    }

    // 下载文件（按 ID）
    public async Task<Stream> DownloadAsync(string fileId, CancellationToken ct = default)
    {
        var stream = new MemoryStream();
        await gridFs.DownloadToStreamAsync(ObjectId.Parse(fileId), stream, cancellationToken: ct);
        stream.Position = 0;
        return stream;
    }

    // 下载文件（按文件名，取最新版本）
    public async Task<Stream> DownloadByNameAsync(string fileName, CancellationToken ct = default)
    {
        var stream = new MemoryStream();
        await gridFs.DownloadToStreamByNameAsync(fileName, stream, cancellationToken: ct);
        stream.Position = 0;
        return stream;
    }

    // 删除文件
    public async Task DeleteAsync(string fileId, CancellationToken ct = default)
        => await gridFs.DeleteAsync(ObjectId.Parse(fileId), ct);

    // 查询文件信息
    public async Task<GridFSFileInfo?> FindInfoAsync(string fileId, CancellationToken ct = default)
    {
        var filter = Builders<GridFSFileInfo>.Filter.Eq("_id", ObjectId.Parse(fileId));
        using var cursor = await gridFs.FindAsync(filter, cancellationToken: ct);
        return await cursor.FirstOrDefaultAsync(ct);
    }
}

builder.Services.AddHealthChecks()
    .AddMongoHealthCheck<MyDbContext>(
        name: "mongodb",                        // 健康检查名称（默认 "mongodb"）
        failureStatus: HealthStatus.Unhealthy,   // 失败状态（默认 Unhealthy）
        tags: ["db", "mongodb"],              // 可选标签（用于分组过滤）
        timeout: TimeSpan.FromSeconds(5));       // 超时时间

var app = builder.Build();

// 暴露统一健康检查端点
app.MapHealthChecks("/health");

// Kubernetes：就绪探针（只检查 DB 连通性）
app.MapHealthChecks("/health/ready", new HealthCheckOptions
{
    Predicate = check => check.Tags.Contains("db")
});

// Kubernetes：存活探针（只检查进程存活）
app.MapHealthChecks("/health/live", new HealthCheckOptions
{
    Predicate = _ => false
});

dotnet add package EasilyNET.AutoDependencyInjection

using EasilyNET.AutoDependencyInjection.Contexts;
using EasilyNET.AutoDependencyInjection.Modules;
using EasilyNET.Mongo.AspNetCore.Serializers;
using EasilyNET.Mongo.ConsoleDebug.Subscribers;
using WebApi.Test.Unit.Common;

internal sealed class MongoModule : AppModule
{
    public override void ConfigureServices(ConfigureServicesContext context)
    {
        var config = context.Configuration;
        var env = context.Environment ?? throw new("获取环境信息出错");

        context.Services.AddMongoContext<DbContext>(config, c =>
        {
            c.DatabaseName = "easilynet";
            c.ClientSettings = cs =>
            {
                cs.ClusterConfigurator = s =>
                {
                    if (env.IsDevelopment())
                    {
                        s.Subscribe(new ActivityEventConsoleDebugSubscriber(new()
                        {
                            Enable = false
                        }));
                    }
                    s.Subscribe(new ActivityEventDiagnosticsSubscriber(new()
                    {
                        CaptureCommandText = true
                    }));
                };
                cs.ApplicationName = Constant.InstanceName;
            };
        });

        context.Services.AddMongoContext<DbContext2>(config, c =>
        {
            c.DatabaseName = "easilynet2";
            c.ClientSettings = cs =>
            {
                cs.ClusterConfigurator = cb => cb.Subscribe(new ActivityEventConsoleDebugSubscriber(new()
                {
                    Enable = false
                }));
                cs.ApplicationName = Constant.InstanceName;
            };
        });

        // 序列化器
        context.Services.RegisterSerializer(new DateOnlySerializerAsString());
        context.Services.RegisterSerializer(new TimeOnlySerializerAsString());
        context.Services.RegisterSerializer(new JsonNodeSerializer());
        context.Services.RegisterSerializer(new JsonObjectSerializer());
        context.Services.RegisterDynamicSerializer();
        context.Services.RegisterGlobalEnumKeyDictionarySerializer();
    }

    public override async Task ApplicationInitialization(ApplicationContext context)
    {
        var app = context.GetApplicationHost() as IApplicationBuilder;
        app?.UseCreateMongoIndexes<DbContext>();
        app?.UseCreateMongoIndexes<DbContext2>();
        await base.ApplicationInitialization(context);
    }
}

[DependsOn(
    typeof(DependencyAppModule),
    typeof(ResponseCompressionModule),
    typeof(CorsModule),
    typeof(ControllersModule),
    typeof(GarnetDistributedCacheModule),
    typeof(MongoModule)
    // ... 其他模块按中间件顺序继续追加
)]
internal sealed class AppWebModule : AppModule
{
    public override void ConfigureServices(ConfigureServicesContext context)
    {
        context.Services.AddProblemDetails();
        context.Services.AddExceptionHandler<BusinessExceptionHandler>();
        context.Services.AddHttpContextAccessor();
    }

    public override async Task ApplicationInitialization(ApplicationContext context)
    {
        var app = context.GetApplicationHost() as IApplicationBuilder;
        app?.UseExceptionHandler();
        app?.UseResponseTime();
        app?.UseAuthentication();
        app?.UseAuthorization();
        app?.UseStaticFiles();
        await base.ApplicationInitialization(context);
    }
}

var builder = WebApplication.CreateBuilder(args);
builder.Services.AddApplicationModules<AppWebModule>();

var app = builder.Build();
app.InitializeApplication();
app.MapControllers();
app.MapHealthChecks("/health");
app.Run();

原因	解决方式
网络不可达 / 防火墙拦截	确认应用机器能访问 `host:port`，安全组已放行
认证或 TLS 错误	检查用户名、密码、`authSource`、`tls` 参数
单节点 / 代理访问	连接串加 `directConnection=true` 或 `loadBalanced=true`
副本集名称不匹配	连接串中的 `replicaSet` 必须与服务端一致
连接池耗尽	降低 `MinConnectionPoolSize`，合理设置 `MaxConnectionPoolSize`

mongodb://user:pwd@host:27017/db?serverSelectionTimeoutMS=5000&connectTimeoutMS=5000&socketTimeoutMS=30000

docker compose -f docker-compose.mongo.rs.yml up -d

joes_du/EasilyNET.Mongo.AspNetCorev6.26.227.92

Get Started

Readme

EasilyNET.Mongo.AspNetCore

目录

安装

⚠️ 中断性变更（Breaking Changes）

不再注册 IMongoClient 和 IMongoDatabase 到 DI 容器

Convention 配置方式变更

快速开始：注册 MongoContext

方式 1：使用 IConfiguration（推荐）

方式 2：使用连接字符串

方式 3：使用 MongoClientSettings

弹性连接配置

字段映射与命名约定

自定义序列化器

DateOnly / TimeOnly

动态 / JSON 类型

枚举键字典

其他

自动创建索引

索引管理策略

自动创建时序集合

自动创建固定大小集合

Atlas Search / Vector Search 索引自动创建

变更流（Change Stream）

1. 定义变更流处理器

2. 注册处理器

断点续传工作原理

GridFS 文件存储

注册 GridFS

键控注入（多 GridFS 桶场景）

使用 GridFS

健康检查

使用 EasilyNET.AutoDependencyInjection 集成

安装模块化 DI

创建 Mongo 模块

创建根模块

Program.cs

常见问题排查

MongoConnectionPoolPausedException: "The connection pool is in paused state"

变更流报错 "Change stream not supported on Standalone"

Atlas Search 索引未创建

不再注册 `IMongoClient` 和 `IMongoDatabase` 到 DI 容器