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);
}
}
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
关键细节:
- 锁竞争窗口:
TryAcquireStartupLockAsync用CreateLinkedTokenSource挂到宿主取消令牌上,再CancelAfter(3s)。若 3 秒内抢不到锁抛OperationCanceledException(且非宿主真正取消)则返回null静默跳过;若抢到则await using确保方法结束时释放锁。 - 节流窗口:抢到锁后仍要读
LastSuccessUtc缓存(键Dpz.Core.MongoIndexInitializer.LastSuccessUtc)。因为缓存底层走 Redis + Backplane,所以多实例共享,能避免 6 小时内多次重启造成重复建索引;只有全部成功才写入该时间(TTL 7 天),有失败则下次启动仍会重试。 - 实体发现:
GetExportedTypes()过滤出IsClass && !IsAbstract且实现的接口之一是泛型IIndexedEntity<>(用GetGenericTypeDefinition()精确比对)的类型列表。 - 逐类型容错:每个实体单独 try/catch,单个失败仅
failedCount++并记错误日志,不中断整批;底层MongoIndexInitializer失败时也会把自己从进程内字典移除,便于重试。 - StopAsync:直接调用
mongoIndexInitializer.ResetAsync清空进程内“已初始化类型”缓存,用于热重启/测试场景下允许重新同步。
四、需要留意的点
Assembly.Load("Dpz.Core.Public.Entity")是按程序集名动态加载,若实体程序集改名或未随部署带上会整体跳过(代码中已做 try/catch 与日志兜底)。- 若配置缺少
mongodb连接串,服务会直接跳过,不抛异常。 - 实例标识
_instanceId为每次启动随机 Guid,仅用于日志区分不同实例。
五、总结
一句话概括:MongoIndexInitializerService 是一个带分布式互斥与跨实例节流保护的启动钩子,负责把实体代码中声明的 MongoDB 索引声明在启动时“对齐”到真实数据库,而把具体“比对/创建/删除/重建索引”的脏活下放给 MongoIndexInitializer.cs。整个流程“锁 → 配置/程序集检查 → 缓存节流 → 反射扫描 → 批量同步 → 成功标记”组成一条清晰的防御性启动链路,与本仓库其它 HostedService(如 AssetUpload、PageRouteCatalogScanner)属于同一套设计模式。