using Dpz.Core.Entity.Base.Indexes;
using Medallion.Threading;
using Microsoft.Extensions.Hosting;

namespace Dpz.Core.Service;

/// <summary>
/// 应用启动时自动扫描所有 IIndexedEntity 实现并创建 MongoDB 索引
/// </summary>
internal sealed class MongoIndexInitializerService(
    IConfiguration configuration,
    IMongoIndexInitializer mongoIndexInitializer,
    IFusionCache fusionCache,
    IDistributedLockProvider distributedLockProvider,
    ILogger<MongoIndexInitializerService> logger
) : IHostedService
{
    private readonly string _instanceId = Guid.NewGuid().ToString();
    private const string LockKey = "Dpz.Core.MongoIndexInitializer.Startup";
    private const string LastSuccessCacheKey = "Dpz.Core.MongoIndexInitializer.LastSuccessUtc";
    private const int DefaultSkipHours = 6;
    private static readonly TimeSpan LockWaitTimeout = TimeSpan.FromSeconds(3);
    private static readonly TimeSpan LastSuccessCacheTtl = TimeSpan.FromDays(7);

    public async Task StartAsync(CancellationToken cancellationToken)
    {
        await using var lockHandle = await TryAcquireStartupLockAsync(cancellationToken);
        if (lockHandle == null)
        {
            logger.LogInformation(
                "实例 {InstanceId} 未获取到索引初始化锁,跳过本次初始化",
                _instanceId
            );
            return;
        }

        var connectionString = configuration.GetConnectionString("mongodb");
        if (string.IsNullOrEmpty(connectionString))
        {
            logger.LogWarning("未配置 mongodb 连接字符串,跳过索引初始化");
            return;
        }

        Assembly entityAssembly;

        try
        {
            entityAssembly = Assembly.Load("Dpz.Core.Public.Entity");
        }
        catch (Exception e)
        {
            logger.LogError(e, "未找到 Dpz.Core.Public.Entity 程序集,忽略初始化索引");
            return;
        }

        logger.LogInformation("实例 {InstanceId} 获取到索引初始化锁,开始初始化", _instanceId);

        var lastSuccess = await fusionCache.TryGetAsync<DateTimeOffset>(
            LastSuccessCacheKey,
            token: cancellationToken
        );
        if (lastSuccess.HasValue)
        {
            var elapsed = DateTimeOffset.UtcNow - lastSuccess.Value;
            var threshold = TimeSpan.FromHours(DefaultSkipHours);
            if (elapsed < threshold)
            {
                logger.LogInformation(
                    "实例 {InstanceId} 跳过索引初始化:距离上次成功仅 {ElapsedMinutes:F1} 分钟,阈值 {SkipHours} 小时",
                    _instanceId,
                    elapsed.TotalMinutes,
                    DefaultSkipHours
                );
                return;
            }
        }

        var stopwatch = Stopwatch.StartNew();

        var indexedTypes = entityAssembly
            .GetExportedTypes()
            .Where(t =>
                t is { IsClass: true, IsAbstract: false }
                && t.GetInterfaces()
                    .Any(i =>
                        i.IsGenericType && i.GetGenericTypeDefinition() == typeof(IIndexedEntity<>)
                    )
            )
            .ToList();

        if (indexedTypes.Count == 0)
        {
            logger.LogInformation("未找到实现 IIndexedEntity 的实体类型");
            return;
        }

        var failedCount = 0;
        foreach (var entityType in indexedTypes)
        {
            try
            {
                await mongoIndexInitializer.SynchronizeForTypeAsync(
                    entityType,
                    connectionString,
                    cancellationToken
                );
            }
            catch (Exception ex)
            {
                failedCount++;
                logger.LogError(ex, "初始化 {EntityType} 的索引时出错", entityType.Name);
            }
        }

        stopwatch.Stop();

        if (failedCount == 0)
        {
            await fusionCache.SetAsync(
                LastSuccessCacheKey,
                DateTimeOffset.UtcNow,
                LastSuccessCacheTtl,
                token: cancellationToken
            );
        }

        logger.LogInformation(
            "MongoDB 索引初始化完成,共处理 {Count} 个实体类型,失败 {FailedCount},耗时 {ElapsedMs}ms",
            indexedTypes.Count,
            failedCount,
            stopwatch.ElapsedMilliseconds
        );
    }

    private async Task<IDistributedSynchronizationHandle?> TryAcquireStartupLockAsync(
        CancellationToken cancellationToken
    )
    {
        using var lockCancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(
            cancellationToken
        );
        lockCancellationTokenSource.CancelAfter(LockWaitTimeout);

        try
        {
            return await distributedLockProvider.AcquireLockAsync(
                LockKey,
                timeout: null,
                cancellationToken: lockCancellationTokenSource.Token
            );
        }
        catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
        {
            return null;
        }
    }

    public Task StopAsync(CancellationToken cancellationToken)
    {
        return mongoIndexInitializer.ResetAsync(cancellationToken);
    }
}
⚠⚠    以下内容为AI分析的结果,请根据实际情况进行判断。

MongoIndexInitializerService.cs 解读报告

一、定位与职责

该文件位于 src/Dpz.Core.Service/MongoIndexInitializerService.cs,是应用启动时的 MongoDB 索引初始化后台服务。它实现 IHostedService,在宿主启动阶段被调用,用于:

  • 自动扫描实体程序集中所有实现了 IIndexedEntity<> 接口的实体类;
  • 依次为这些实体对应的 MongoDB 集合同步索引(缺则建、漂移则删了重建);
  • 通过分布式锁保证多实例部署时只有一个实例执行;
  • 通过 FusionCache 记录“最近一次成功初始化时间”,默认 6 小时内的后续启动直接跳过,避免重复劳动。

它本身不做索引的具体创建逻辑,真正的同步实现委托给 src/Dpz.Core.MongodbAccess/MongoIndexInitializer.cs(接口定义在 IMongoIndexInitializer.cs)。

二、依赖注入与协作关系

依赖全部通过主构造函数注入,并在 ServiceExtensions.cs 中注册为 AddHostedService<MongoIndexInitializerService>(),同时注册单例 IMongoIndexInitializer → MongoIndexInitializer。

注入依赖用途
IConfiguration读取 ConnectionStrings:mongodb 连接串
IMongoIndexInitializer执行具体集合索引同步(SynchronizeForTypeAsync)、停止时 ResetAsync
IFusionCache记录/读取上次成功时间(L1/L2 + Redis Backplane,见 AddFusionCacheServices)
IDistributedLockProvider获取分布式锁(DEBUG 为文件锁,Release 为 Redis 锁,见 ServiceExtensions)
ILogger<...>日志

协作对象还包括:

  • 动态加载的程序集 Dpz.Core.Public.Entity(对应目录 src/Dpz.Core.Public.Entity),通过 Assembly.Load 按名字加载——这是运行时反射解耦,Service 工程并不静态引用它;
  • 索引契约接口 IIndexedEntity.cs:泛型接口,用 C# 11 static abstract 声明 GetIndexDefinitions(),实体通过实现它来声明索引定义;
  • 底层执行器 MongoIndexInitializer.cs:用 ConcurrentDictionary<Type, Task> 保证进程内每个类型只同步一次,失败时 TryRemove 允许重试;内部通过反射 MakeGenericMethod 调用泛型核心逻辑,并比对已存在的索引(名称、key 方向、unique、TTL、文本权重等),不一致时 Drop 后重建。

说明:Dpz.Core.Public.Entity 项目内部是否有实体实现了 IIndexedEntity<>、具体有哪些,本次未逐一展开核对。

三、启动流程(StartAsync)

该流程在同目录的 README.md 中也有简要描述,且与 AssetUploadHostedService.cs、PageRouteCatalogScannerService.cs 采用一致的“分布式锁 + 上次成功缓存 + 反射批量处理”模板。

flowchart TD
    A[宿主启动: StartAsync] --> B[尝试获取分布式锁 LockKey<br/>等待上限 3 秒]
    B -- 拿不到锁 --> C[记录日志: 本实例跳过初始化] --> Z[结束]
    B -- 拿到锁 --> D{配置中有 mongodb<br/>连接字符串?}
    D -- 无 --> E[告警日志并跳过]
    D -- 有 --> F{Assembly.Load<br/>Dpz.Core.Public.Entity 成功?}
    F -- 失败 --> G[错误日志, 忽略索引]
    F -- 成功 --> H[读 FusionCache 的 LastSuccessUtc]
    H -- 存在且距今 < 6 小时 --> I[日志: 跳过重复初始化] --> Z
    H -- 无记录或超过 6 小时 --> J[反射扫描所有非抽象类<br/>且实现 IIndexedEntity<> 的实体]
    J -- 数量为 0 --> K[日志: 未找到实体] --> Z
    J -- 逐个调用 --> L[mongoIndexInitializer.SynchronizeForTypeAsync<br/>统计失败数, 单个失败不影响其它]
    L --> M{failedCount == 0?}
    M -- 是 --> N[FusionCache 写入当前 UTC 时间<br/>TTL 7 天]
    M -- 否 --> O
    N --> O[输出汇总日志: 数量/失败数/耗时]
    O --> Z

    subgraph 停止阶段
        Z2[宿主停止: StopAsync] --> P[mongoIndexInitializer.ResetAsync<br/>清空进程内已初始化类型缓存]
    end

关键细节:

  1. 锁竞争窗口:TryAcquireStartupLockAsync 用 CreateLinkedTokenSource 挂到宿主取消令牌上,再 CancelAfter(3s)。若 3 秒内抢不到锁抛 OperationCanceledException(且非宿主真正取消)则返回 null 静默跳过;若抢到则 await using 确保方法结束时释放锁。
  2. 节流窗口:抢到锁后仍要读 LastSuccessUtc 缓存(键 Dpz.Core.MongoIndexInitializer.LastSuccessUtc)。因为缓存底层走 Redis + Backplane,所以多实例共享,能避免 6 小时内多次重启造成重复建索引;只有全部成功才写入该时间(TTL 7 天),有失败则下次启动仍会重试。
  3. 实体发现:GetExportedTypes() 过滤出 IsClass && !IsAbstract 且实现的接口之一是泛型 IIndexedEntity<>(用 GetGenericTypeDefinition() 精确比对)的类型列表。
  4. 逐类型容错:每个实体单独 try/catch,单个失败仅 failedCount++ 并记错误日志,不中断整批;底层 MongoIndexInitializer 失败时也会把自己从进程内字典移除,便于重试。
  5. StopAsync:直接调用 mongoIndexInitializer.ResetAsync 清空进程内“已初始化类型”缓存,用于热重启/测试场景下允许重新同步。

四、需要留意的点

  • Assembly.Load("Dpz.Core.Public.Entity") 是按程序集名动态加载,若实体程序集改名或未随部署带上会整体跳过(代码中已做 try/catch 与日志兜底)。
  • 若配置缺少 mongodb 连接串,服务会直接跳过,不抛异常。
  • 实例标识 _instanceId 为每次启动随机 Guid,仅用于日志区分不同实例。

五、总结

一句话概括:MongoIndexInitializerService 是一个带分布式互斥与跨实例节流保护的启动钩子,负责把实体代码中声明的 MongoDB 索引声明在启动时“对齐”到真实数据库,而把具体“比对/创建/删除/重建索引”的脏活下放给 MongoIndexInitializer.cs。整个流程“锁 → 配置/程序集检查 → 缓存节流 → 反射扫描 → 批量同步 → 成功标记”组成一条清晰的防御性启动链路,与本仓库其它 HostedService(如 AssetUpload、PageRouteCatalogScanner)属于同一套设计模式。

评论加载中...