“教学班成绩分析”页面新增“定时刷新设置”，支持启停、刷新间隔、单次处理上限、查看上次/下次扫描时间。
配置保存在独立数据库表 CourseGradeStatisticsRefreshSettings，不是修改 appsettings.json。 后台每 10 秒读取配置，仅到期扫描；成绩录入、导入、审批时不再立即创建统计任务。 扫描发现过期数据后，仍通过持久任务、Outbox 和 RabbitMQ 执行；Redis统计缓存由处理任务统一刷新。
This commit is contained in:
@@ -0,0 +1,228 @@
|
||||
using Jiaowu.Api.Domain.Academic;
|
||||
using Jiaowu.Api.Domain.System;
|
||||
using Jiaowu.Api.Infrastructure.BackgroundJobs;
|
||||
using Jiaowu.Api.Infrastructure.Persistence;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
|
||||
namespace Jiaowu.Api.Infrastructure.Grades;
|
||||
|
||||
/// <summary>
|
||||
/// Repairs missing or stale materialized grade statistics on a database-
|
||||
/// configured fixed interval. Grade writes do not enqueue refresh jobs; this
|
||||
/// worker batches changes made during bulk imports.
|
||||
/// </summary>
|
||||
public sealed class CourseGradeStatisticsRefreshWorker(
|
||||
IServiceScopeFactory scopeFactory,
|
||||
TimeProvider timeProvider,
|
||||
ILogger<CourseGradeStatisticsRefreshWorker> logger) : BackgroundService
|
||||
{
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
logger.LogInformation("Database-configured course grade statistics scheduler started.");
|
||||
using var timer = new PeriodicTimer(TimeSpan.FromSeconds(10), timeProvider);
|
||||
while (!stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
try
|
||||
{
|
||||
await using var scope = scopeFactory.CreateAsyncScope();
|
||||
var scheduler = scope.ServiceProvider
|
||||
.GetRequiredService<CourseGradeStatisticsRefreshScheduler>();
|
||||
var queued = await scheduler.EnqueueDueAsync(
|
||||
timeProvider.GetUtcNow().UtcDateTime,
|
||||
stoppingToken);
|
||||
if (queued > 0)
|
||||
logger.LogInformation(
|
||||
"Scheduled course grade statistics scan queued {Count} refresh jobs.",
|
||||
queued);
|
||||
}
|
||||
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
break;
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
logger.LogError(exception, "Scheduled course grade statistics scan failed.");
|
||||
}
|
||||
|
||||
if (!await timer.WaitForNextTickAsync(stoppingToken)) break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public sealed class CourseGradeStatisticsRefreshScheduler(
|
||||
AppDbContext db,
|
||||
ILogger<CourseGradeStatisticsRefreshScheduler> logger)
|
||||
{
|
||||
public async Task<int> EnqueueDueAsync(
|
||||
DateTime utcNow,
|
||||
CancellationToken cancellationToken) =>
|
||||
await ExecuteWithLeaseAsync(async () =>
|
||||
{
|
||||
var setting = await db.CourseGradeStatisticsRefreshSettings
|
||||
.SingleOrDefaultAsync(
|
||||
x => x.Key == CourseGradeStatisticsRefreshSetting.DefaultKey,
|
||||
cancellationToken);
|
||||
if (setting is null)
|
||||
{
|
||||
setting = new CourseGradeStatisticsRefreshSetting();
|
||||
db.CourseGradeStatisticsRefreshSettings.Add(setting);
|
||||
}
|
||||
|
||||
var interval = TimeSpan.FromSeconds(
|
||||
Math.Clamp(setting.IntervalSeconds, 10, 86400));
|
||||
if (!setting.IsEnabled ||
|
||||
setting.LastRunAt.HasValue && utcNow < setting.LastRunAt.Value + interval)
|
||||
{
|
||||
if (db.Entry(setting).State == EntityState.Added)
|
||||
await db.SaveChangesAsync(cancellationToken);
|
||||
return 0;
|
||||
}
|
||||
|
||||
setting.LastRunAt = utcNow;
|
||||
var queued = await EnqueueStaleCoreAsync(setting.BatchSize, cancellationToken);
|
||||
await db.SaveChangesAsync(cancellationToken);
|
||||
return queued;
|
||||
}, cancellationToken);
|
||||
|
||||
public async Task<int> EnqueueStaleAsync(
|
||||
int batchSize,
|
||||
CancellationToken cancellationToken) =>
|
||||
await ExecuteWithLeaseAsync(async () =>
|
||||
{
|
||||
var queued = await EnqueueStaleCoreAsync(batchSize, cancellationToken);
|
||||
if (queued > 0) await db.SaveChangesAsync(cancellationToken);
|
||||
return queued;
|
||||
}, cancellationToken);
|
||||
|
||||
private async Task<int> ExecuteWithLeaseAsync(
|
||||
Func<Task<int>> action,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var usesMySqlLease = db.Database.ProviderName?.Contains(
|
||||
"MySql",
|
||||
StringComparison.OrdinalIgnoreCase) == true;
|
||||
if (usesMySqlLease && !await TryAcquireMySqlLeaseAsync(cancellationToken))
|
||||
{
|
||||
await db.Database.CloseConnectionAsync();
|
||||
logger.LogDebug("Another instance owns the grade statistics refresh lease.");
|
||||
return 0;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
return await action();
|
||||
}
|
||||
finally
|
||||
{
|
||||
if (usesMySqlLease)
|
||||
await ReleaseMySqlLeaseAsync();
|
||||
}
|
||||
}
|
||||
|
||||
private async Task<int> EnqueueStaleCoreAsync(
|
||||
int batchSize,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
batchSize = Math.Clamp(batchSize, 1, 5000);
|
||||
var activeTargets = await (
|
||||
from job in db.CourseGradeStatisticsRefreshJobs.AsNoTracking()
|
||||
join sheet in db.GradeSheets.AsNoTracking()
|
||||
on job.GradeSheetId equals sheet.Id
|
||||
where job.Status == CourseGradeStatisticsRefreshJobStatus.Queued ||
|
||||
job.Status == CourseGradeStatisticsRefreshJobStatus.Running
|
||||
select new CourseTermTarget(
|
||||
sheet.TeachingTask!.CourseId,
|
||||
sheet.TeachingTask.AcademicTermId))
|
||||
.Distinct()
|
||||
.ToListAsync(cancellationToken);
|
||||
var active = activeTargets.ToHashSet();
|
||||
|
||||
var rows = await db.GradeSheets.AsNoTracking()
|
||||
.Where(sheet =>
|
||||
sheet.Status == GradeSheetStatus.Published &&
|
||||
sheet.Records.Any(record => record.TotalScore != null))
|
||||
.Select(sheet => new RefreshCandidate(
|
||||
sheet.Id,
|
||||
sheet.TeachingTask!.CourseId,
|
||||
sheet.TeachingTask.AcademicTermId,
|
||||
sheet.UpdatedAt,
|
||||
sheet.Records
|
||||
.Where(record => record.TotalScore != null)
|
||||
.Max(record => record.UpdatedAt),
|
||||
db.TeachingTaskGradeStatistics
|
||||
.Where(statistic => statistic.GradeSheetId == sheet.Id)
|
||||
.Select(statistic => (DateTime?)statistic.CalculatedAt)
|
||||
.FirstOrDefault()))
|
||||
.ToListAsync(cancellationToken);
|
||||
|
||||
var stale = rows
|
||||
.Where(row =>
|
||||
row.CalculatedAt is null ||
|
||||
row.SheetUpdatedAt > row.CalculatedAt ||
|
||||
row.RecordsUpdatedAt > row.CalculatedAt)
|
||||
.GroupBy(row => new CourseTermTarget(row.CourseId, row.AcademicTermId))
|
||||
.Where(group => !active.Contains(group.Key))
|
||||
.Select(group => group
|
||||
.OrderByDescending(row => row.RecordsUpdatedAt)
|
||||
.ThenByDescending(row => row.SheetUpdatedAt)
|
||||
.First())
|
||||
.Take(batchSize)
|
||||
.ToArray();
|
||||
|
||||
foreach (var candidate in stale)
|
||||
{
|
||||
var job = new CourseGradeStatisticsRefreshJob
|
||||
{
|
||||
GradeSheetId = candidate.GradeSheetId
|
||||
};
|
||||
db.CourseGradeStatisticsRefreshJobs.Add(job);
|
||||
db.BackgroundJobOutboxMessages.Add(BackgroundJobOutboxMessage.Create(
|
||||
BackgroundJobKind.CourseGradeStatisticsRefresh,
|
||||
job.Id));
|
||||
}
|
||||
|
||||
if (stale.Length == 0) return 0;
|
||||
logger.LogDebug(
|
||||
"Queued {Count} stale course grade statistics targets.",
|
||||
stale.Length);
|
||||
return stale.Length;
|
||||
}
|
||||
|
||||
private async Task<bool> TryAcquireMySqlLeaseAsync(
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
await db.Database.OpenConnectionAsync(cancellationToken);
|
||||
await using var command = db.Database.GetDbConnection().CreateCommand();
|
||||
command.CommandText = "SELECT GET_LOCK('jiaowu:grade-statistics-refresh', 0);";
|
||||
var result = await command.ExecuteScalarAsync(cancellationToken);
|
||||
return Convert.ToInt32(result) == 1;
|
||||
}
|
||||
|
||||
private async Task ReleaseMySqlLeaseAsync()
|
||||
{
|
||||
try
|
||||
{
|
||||
await using var command = db.Database.GetDbConnection().CreateCommand();
|
||||
command.CommandText = "SELECT RELEASE_LOCK('jiaowu:grade-statistics-refresh');";
|
||||
await command.ExecuteScalarAsync(CancellationToken.None);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
logger.LogWarning(exception, "Failed to release grade statistics refresh lease.");
|
||||
}
|
||||
finally
|
||||
{
|
||||
await db.Database.CloseConnectionAsync();
|
||||
}
|
||||
}
|
||||
|
||||
private sealed record RefreshCandidate(
|
||||
Guid GradeSheetId,
|
||||
Guid CourseId,
|
||||
Guid AcademicTermId,
|
||||
DateTime SheetUpdatedAt,
|
||||
DateTime RecordsUpdatedAt,
|
||||
DateTime? CalculatedAt);
|
||||
|
||||
private sealed record CourseTermTarget(Guid CourseId, Guid AcademicTermId);
|
||||
}
|
||||
Reference in New Issue
Block a user