第二阶段优化已完成，重点是提高后台任务吞吐并降低 Outbox 数据库开销。

RabbitMQ 默认从单一回调通道提升为三类任务独立并行，跨类型并发能力由 1 提升到 3；每类还能独立配置 1–16 个消费者。[RabbitMqBackgroundJobs.cs (line 143)](E:/jiaowu/src/Jiaowu.Api/Infrastructure/BackgroundJobs/RabbitMqBackgroundJobs.cs:143) InMemory 开发模式同步改为按任务类型隔离队列，避免某类长任务堵塞其他任务。租约恢复检查从“每发布一条执行一次”改为默认每 60 秒维护一次。启动恢复改为数据库 NOT EXISTS 查询，不再把全部历史 Outbox 加载进内存。[BackgroundJobOutboxPublisher.cs (line 9)](E:/jiaowu/src/Jiaowu.Api/Infrastructure/BackgroundJobs/BackgroundJobOutboxPublisher.cs:9) 完成消息默认保留 14 天，之后按每批 500 条清理，并增加对应组合索引。 /health/messaging 现在返回各状态积压量、过期租约和最老任务等待时间。[BackgroundJobMonitoringService.cs (line 16)](E:/jiaowu/src/Jiaowu.Api/Infrastructure/BackgroundJobs/BackgroundJobMonitoringService.cs:16) 新增 Jiaowu.BackgroundJobs 运行时指标，覆盖发布量、处理量、发布耗时、处理耗时和清理量。[BackgroundJobTelemetry.cs (line 9)](E:/jiaowu/src/Jiaowu.Api/Infrastructure/BackgroundJobs/BackgroundJobTelemetry.cs:9) 配置、Compose 和调优建议已更新。[README.md (line 257)](E:/jiaowu/README.md:257)
2026-07-26 21:05:18 +08:00
parent 6300645120
commit 034f1eacb8
21 changed files with 579 additions and 135 deletions
@@ -0,0 +1,56 @@
+using Jiaowu.Api.Domain.System;
+using Jiaowu.Api.Infrastructure.Persistence;
+using Microsoft.EntityFrameworkCore;
+
+namespace Jiaowu.Api.Infrastructure.BackgroundJobs;
+
+public sealed record BackgroundJobBacklogSnapshot(
+    int Pending,
+    int Publishing,
+    int Published,
+    int Processing,
+    int ExpiredLeases,
+    DateTime? OldestUnfinishedAt,
+    double? OldestUnfinishedAgeSeconds);
+
+public sealed class BackgroundJobMonitoringService(AppDbContext db)
+{
+    public async Task<BackgroundJobBacklogSnapshot> GetSnapshotAsync(
+        CancellationToken cancellationToken)
+    {
+        var counts = await db.BackgroundJobOutboxMessages.AsNoTracking()
+            .Where(x => x.State != BackgroundJobOutboxState.Completed)
+            .GroupBy(x => x.State)
+            .Select(group => new { State = group.Key, Count = group.Count() })
+            .ToListAsync(cancellationToken);
+        var countByState = counts.ToDictionary(x => x.State, x => x.Count);
+        var now = DateTime.UtcNow;
+        var expiredLeases = await db.BackgroundJobOutboxMessages.AsNoTracking()
+            .CountAsync(
+                x => (x.State == BackgroundJobOutboxState.Publishing ||
+                      x.State == BackgroundJobOutboxState.Processing) &&
+                     x.LeaseExpiresAt != null &&
+                     x.LeaseExpiresAt < now,
+                cancellationToken);
+        var oldestUnfinishedAt = await db.BackgroundJobOutboxMessages
+            .AsNoTracking()
+            .Where(x => x.State != BackgroundJobOutboxState.Completed)
+            .Select(x => (DateTime?)x.CreatedAt)
+            .MinAsync(cancellationToken);
+        double? ageSeconds = oldestUnfinishedAt.HasValue
+            ? Math.Max(0, (now - oldestUnfinishedAt.Value).TotalSeconds)
+            : null;
+
+        return new BackgroundJobBacklogSnapshot(
+            GetCount(BackgroundJobOutboxState.Pending),
+            GetCount(BackgroundJobOutboxState.Publishing),
+            GetCount(BackgroundJobOutboxState.Published),
+            GetCount(BackgroundJobOutboxState.Processing),
+            expiredLeases,
+            oldestUnfinishedAt,
+            ageSeconds);
+
+        int GetCount(BackgroundJobOutboxState state) =>
+            countByState.GetValueOrDefault(state);
+    }
+}
@@ -1,3 +1,5 @@
+using Jiaowu.Api.Domain.System;
+
 namespace Jiaowu.Api.Infrastructure.BackgroundJobs;

 public sealed class BackgroundJobOptions
@@ -8,13 +10,27 @@ public sealed class BackgroundJobOptions
    public int PollIntervalMilliseconds { get; set; } = 500;
    public int LeaseSeconds { get; set; } = 120;
    public ushort PrefetchCount { get; set; } = 1;
+    public int AutomaticScheduleConcurrency { get; set; } = 1;
+    public int SchedulePublishConcurrency { get; set; } = 1;
+    public int MakeupExamAutoConcurrency { get; set; } = 1;
    public string Exchange { get; set; } = "jiaowu.background-jobs";
    public string QueuePrefix { get; set; } = "jiaowu.background-jobs";
    public bool UseQuorumQueues { get; set; } = true;
    public int ProcessingAttemptLimit { get; set; } = 5;
+    public int MaintenanceIntervalSeconds { get; set; } = 60;
+    public int CompletedRetentionDays { get; set; } = 14;
+    public int CleanupBatchSize { get; set; } = 500;

    public bool UsesRabbitMq =>
        Transport.Equals("RabbitMq", StringComparison.OrdinalIgnoreCase);
+
+    public int ConsumerConcurrency(BackgroundJobKind kind) => kind switch
+    {
+        BackgroundJobKind.AutomaticSchedule => AutomaticScheduleConcurrency,
+        BackgroundJobKind.SchedulePublish => SchedulePublishConcurrency,
+        BackgroundJobKind.MakeupExamAuto => MakeupExamAutoConcurrency,
+        _ => throw new ArgumentOutOfRangeException(nameof(kind), kind, null)
+    };
 }

 public sealed class RabbitMqOptions
@@ -1,3 +1,4 @@
+using System.Diagnostics;
 using Jiaowu.Api.Domain.Academic;
 using Jiaowu.Api.Domain.System;
 using Jiaowu.Api.Infrastructure.Persistence;
@@ -9,17 +10,26 @@ public sealed class BackgroundJobOutboxPublisher(
    IServiceScopeFactory scopeFactory,
    IBackgroundJobTransport transport,
    BackgroundJobOptions options,
+    BackgroundJobTelemetry telemetry,
    ILogger<BackgroundJobOutboxPublisher> logger) : BackgroundService
 {
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await PrepareForStartupAsync(stoppingToken);
+        var nextMaintenanceAt = DateTime.MinValue;

        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
-                await RecoverExpiredProcessingAsync(stoppingToken);
+                if (DateTime.UtcNow >= nextMaintenanceAt)
+                {
+                    await RecoverExpiredProcessingAsync(stoppingToken);
+                    await CleanupCompletedAsync(stoppingToken);
+                    nextMaintenanceAt = DateTime.UtcNow.AddSeconds(
+                        options.MaintenanceIntervalSeconds);
+                }
+
                var claimed = await ClaimNextAsync(stoppingToken);
                if (claimed is null)
                {
@@ -62,51 +72,107 @@ public sealed class BackgroundJobOutboxPublisher(
        await AddMissingOutboxMessagesAsync(db, cancellationToken);
    }

-    private static async Task AddMissingOutboxMessagesAsync(
+    private async Task AddMissingOutboxMessagesAsync(
        AppDbContext db,
        CancellationToken cancellationToken)
    {
-        var existing = await db.BackgroundJobOutboxMessages.AsNoTracking()
-            .Select(x => new { x.JobKind, x.JobId })
-            .ToListAsync(cancellationToken);
-        var existingKeys = existing
-            .Select(x => (x.JobKind, x.JobId))
-            .ToHashSet();
-
        var automaticJobs = await db.AutomaticScheduleJobs.AsNoTracking()
-            .Where(x => x.Status == AutomaticScheduleJobStatus.Queued ||
-                        x.Status == AutomaticScheduleJobStatus.Running)
+            .Where(x =>
+                (x.Status == AutomaticScheduleJobStatus.Queued ||
+                 x.Status == AutomaticScheduleJobStatus.Running) &&
+                !db.BackgroundJobOutboxMessages.Any(message =>
+                    message.JobKind == BackgroundJobKind.AutomaticSchedule &&
+                    message.JobId == x.Id))
            .Select(x => x.Id)
            .ToListAsync(cancellationToken);
        var publishJobs = await db.SchedulePublishJobs.AsNoTracking()
-            .Where(x => x.Status == SchedulePublishJobStatus.Queued ||
-                        x.Status == SchedulePublishJobStatus.Running)
+            .Where(x =>
+                (x.Status == SchedulePublishJobStatus.Queued ||
+                 x.Status == SchedulePublishJobStatus.Running) &&
+                !db.BackgroundJobOutboxMessages.Any(message =>
+                    message.JobKind == BackgroundJobKind.SchedulePublish &&
+                    message.JobId == x.Id))
            .Select(x => x.Id)
            .ToListAsync(cancellationToken);
        var makeupJobs = await db.MakeupExamAutoJobs.AsNoTracking()
-            .Where(x => x.Status == MakeupExamAutoJobStatus.Queued ||
-                        x.Status == MakeupExamAutoJobStatus.Running)
+            .Where(x =>
+                (x.Status == MakeupExamAutoJobStatus.Queued ||
+                 x.Status == MakeupExamAutoJobStatus.Running) &&
+                !db.BackgroundJobOutboxMessages.Any(message =>
+                    message.JobKind == BackgroundJobKind.MakeupExamAuto &&
+                    message.JobId == x.Id))
            .Select(x => x.Id)
            .ToListAsync(cancellationToken);

-        AddMissing(BackgroundJobKind.AutomaticSchedule, automaticJobs);
-        AddMissing(BackgroundJobKind.SchedulePublish, publishJobs);
-        AddMissing(BackgroundJobKind.MakeupExamAuto, makeupJobs);
-
-        if (db.ChangeTracker.HasChanges())
-            await db.SaveChangesAsync(cancellationToken);
-
-        void AddMissing(BackgroundJobKind kind, IEnumerable<Guid> jobIds)
+        var missingKeys = automaticJobs
+            .Select(id => (BackgroundJobKind.AutomaticSchedule, id))
+            .Concat(publishJobs.Select(id =>
+                (BackgroundJobKind.SchedulePublish, id)))
+            .Concat(makeupJobs.Select(id =>
+                (BackgroundJobKind.MakeupExamAuto, id)))
+            .ToList();
+        foreach (var (kind, jobId) in missingKeys)
        {
-            foreach (var jobId in jobIds)
-            {
-                if (existingKeys.Add((kind, jobId)))
-                {
-                    db.BackgroundJobOutboxMessages.Add(
-                        BackgroundJobOutboxMessage.Create(kind, jobId));
-                }
-            }
+            db.BackgroundJobOutboxMessages.Add(
+                BackgroundJobOutboxMessage.Create(kind, jobId));
        }
+
+        if (!db.ChangeTracker.HasChanges())
+            return;
+
+        try
+        {
+            await db.SaveChangesAsync(cancellationToken);
+        }
+        catch (DbUpdateException)
+        {
+            db.ChangeTracker.Clear();
+            foreach (var (kind, jobId) in missingKeys)
+            {
+                var recovered = await db.BackgroundJobOutboxMessages
+                    .AsNoTracking()
+                    .AnyAsync(
+                        message => message.JobKind == kind &&
+                                   message.JobId == jobId,
+                        cancellationToken);
+                if (!recovered)
+                    throw;
+            }
+
+            logger.LogInformation(
+                "Another application instance recovered {Count} missing outbox messages first.",
+                missingKeys.Count);
+        }
+    }
+
+    private async Task CleanupCompletedAsync(CancellationToken cancellationToken)
+    {
+        await using var scope = scopeFactory.CreateAsyncScope();
+        var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
+        var cutoff = DateTime.UtcNow.AddDays(-options.CompletedRetentionDays);
+        var completedIds = await db.BackgroundJobOutboxMessages.AsNoTracking()
+            .Where(x =>
+                x.State == BackgroundJobOutboxState.Completed &&
+                x.CompletedAt != null &&
+                x.CompletedAt < cutoff)
+            .OrderBy(x => x.CompletedAt)
+            .Select(x => x.Id)
+            .Take(options.CleanupBatchSize)
+            .ToListAsync(cancellationToken);
+        if (completedIds.Count == 0)
+            return;
+
+        foreach (var id in completedIds)
+        {
+            db.BackgroundJobOutboxMessages.Remove(
+                new BackgroundJobOutboxMessage { Id = id });
+        }
+        await db.SaveChangesAsync(cancellationToken);
+        telemetry.RecordCleaned(completedIds.Count);
+        logger.LogInformation(
+            "Removed {Count} completed background job outbox messages older than {Cutoff}.",
+            completedIds.Count,
+            cutoff);
    }

    private async Task RecoverExpiredProcessingAsync(
@@ -175,6 +241,7 @@ public sealed class BackgroundJobOutboxPublisher(
        Guid token,
        CancellationToken cancellationToken)
    {
+        var startedAt = Stopwatch.GetTimestamp();
        try
        {
            await transport.PublishAsync(message, cancellationToken);
@@ -193,6 +260,11 @@ public sealed class BackgroundJobOutboxPublisher(
                        .SetProperty(x => x.ProcessingToken, (Guid?)null)
                        .SetProperty(x => x.LeaseExpiresAt, (DateTime?)null),
                    cancellationToken);
+            telemetry.RecordPublish(
+                message.JobKind,
+                transport.IsDurable,
+                "published",
+                Stopwatch.GetElapsedTime(startedAt));
        }
        catch (Exception exception)
        {
@@ -213,6 +285,11 @@ public sealed class BackgroundJobOutboxPublisher(
                        .SetProperty(x => x.LeaseExpiresAt, (DateTime?)null)
                        .SetProperty(x => x.LastError, messageText),
                    cancellationToken);
+            telemetry.RecordPublish(
+                message.JobKind,
+                transport.IsDurable,
+                "failed",
+                Stopwatch.GetElapsedTime(startedAt));
            throw;
        }
    }
@@ -1,3 +1,4 @@
+using System.Diagnostics;
 using Jiaowu.Api.Domain.Academic;
 using Jiaowu.Api.Domain.System;
 using Jiaowu.Api.Infrastructure.Exams;
@@ -28,12 +29,14 @@ public sealed class BackgroundJobRunner(
    IServiceScopeFactory scopeFactory,
    IBackgroundJobTransport transport,
    BackgroundJobOptions options,
+    BackgroundJobTelemetry telemetry,
    ILogger<BackgroundJobRunner> logger)
 {
    public async Task<BackgroundJobRunResult> RunAsync(
        BackgroundJobEnvelope message,
        CancellationToken cancellationToken)
    {
+        var startedAt = Stopwatch.GetTimestamp();
        var token = Guid.NewGuid();
        var claim = await TryClaimAsync(message, token, cancellationToken);
        if (!claim.Claimed)
@@ -48,6 +51,10 @@ public sealed class BackgroundJobRunner(
            await MarkJobRetryLimitExceededAsync(message, cancellationToken);
            await MarkCompletedAsync(message.OutboxMessageId, token,
                cancellationToken);
+            telemetry.RecordProcessing(
+                message.JobKind,
+                "retry-limit",
+                Stopwatch.GetElapsedTime(startedAt));
            return BackgroundJobRunResult.Completed;
        }

@@ -85,6 +92,10 @@ public sealed class BackgroundJobRunner(

            await MarkCompletedAsync(message.OutboxMessageId, token,
                cancellationToken);
+            telemetry.RecordProcessing(
+                message.JobKind,
+                "completed",
+                Stopwatch.GetElapsedTime(startedAt));
            return BackgroundJobRunResult.Completed;
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
@@ -103,6 +114,10 @@ public sealed class BackgroundJobRunner(
                token,
                exception,
                CancellationToken.None);
+            telemetry.RecordProcessing(
+                message.JobKind,
+                "retry",
+                Stopwatch.GetElapsedTime(startedAt));
            return BackgroundJobRunResult.Retry(TimeSpan.FromSeconds(2));
        }
        finally
@@ -0,0 +1,74 @@
+using System.Diagnostics;
+using System.Diagnostics.Metrics;
+using Jiaowu.Api.Domain.System;
+
+namespace Jiaowu.Api.Infrastructure.BackgroundJobs;
+
+public sealed class BackgroundJobTelemetry : IDisposable
+{
+    public const string MeterName = "Jiaowu.BackgroundJobs";
+
+    private readonly Meter _meter = new(MeterName, "1.0.0");
+    private readonly Counter<long> _published;
+    private readonly Counter<long> _processed;
+    private readonly Histogram<double> _publishDuration;
+    private readonly Histogram<double> _processingDuration;
+    private readonly Counter<long> _cleaned;
+
+    public BackgroundJobTelemetry()
+    {
+        _published = _meter.CreateCounter<long>(
+            "jiaowu.background_jobs.published",
+            unit: "{message}");
+        _processed = _meter.CreateCounter<long>(
+            "jiaowu.background_jobs.processed",
+            unit: "{job}");
+        _publishDuration = _meter.CreateHistogram<double>(
+            "jiaowu.background_jobs.publish.duration",
+            unit: "ms");
+        _processingDuration = _meter.CreateHistogram<double>(
+            "jiaowu.background_jobs.processing.duration",
+            unit: "ms");
+        _cleaned = _meter.CreateCounter<long>(
+            "jiaowu.background_jobs.outbox.cleaned",
+            unit: "{message}");
+    }
+
+    public void RecordPublish(
+        BackgroundJobKind kind,
+        bool durable,
+        string outcome,
+        TimeSpan duration)
+    {
+        var tags = new TagList
+        {
+            { "job.kind", kind.ToString() },
+            { "messaging.backend", durable ? "rabbitmq" : "memory" },
+            { "job.outcome", outcome }
+        };
+        _published.Add(1, tags);
+        _publishDuration.Record(duration.TotalMilliseconds, tags);
+    }
+
+    public void RecordProcessing(
+        BackgroundJobKind kind,
+        string outcome,
+        TimeSpan duration)
+    {
+        var tags = new TagList
+        {
+            { "job.kind", kind.ToString() },
+            { "job.outcome", outcome }
+        };
+        _processed.Add(1, tags);
+        _processingDuration.Record(duration.TotalMilliseconds, tags);
+    }
+
+    public void RecordCleaned(int count)
+    {
+        if (count > 0)
+            _cleaned.Add(count);
+    }
+
+    public void Dispose() => _meter.Dispose();
+}
@@ -21,26 +21,31 @@ public interface IBackgroundJobTransport

 public sealed class InMemoryBackgroundJobTransport : IBackgroundJobTransport
 {
-    private readonly Channel<BackgroundJobEnvelope> _channel =
-        Channel.CreateBounded<BackgroundJobEnvelope>(
-            new BoundedChannelOptions(256)
-            {
-                FullMode = BoundedChannelFullMode.Wait,
-                SingleReader = true,
-                SingleWriter = false
-            });
+    private readonly IReadOnlyDictionary<
+        BackgroundJobKind,
+        Channel<BackgroundJobEnvelope>> _channels =
+        Enum.GetValues<BackgroundJobKind>().ToDictionary(
+            kind => kind,
+            _ => Channel.CreateBounded<BackgroundJobEnvelope>(
+                new BoundedChannelOptions(256)
+                {
+                    FullMode = BoundedChannelFullMode.Wait,
+                    SingleReader = false,
+                    SingleWriter = false
+                }));

    public bool IsDurable => false;

    public ValueTask PublishAsync(
        BackgroundJobEnvelope message,
        CancellationToken cancellationToken) =>
-        _channel.Writer.WriteAsync(message, cancellationToken);
+        _channels[message.JobKind].Writer.WriteAsync(message, cancellationToken);

    public Task<bool> CheckHealthAsync(CancellationToken cancellationToken) =>
        Task.FromResult(true);

    public IAsyncEnumerable<BackgroundJobEnvelope> ReadAllAsync(
+        BackgroundJobKind kind,
        CancellationToken cancellationToken) =>
-        _channel.Reader.ReadAllAsync(cancellationToken);
+        _channels[kind].Reader.ReadAllAsync(cancellationToken);
 }
@@ -1,27 +1,46 @@
+using Jiaowu.Api.Domain.System;
+
 namespace Jiaowu.Api.Infrastructure.BackgroundJobs;

 public sealed class InMemoryBackgroundJobWorker(
    InMemoryBackgroundJobTransport transport,
    BackgroundJobRunner runner,
+    BackgroundJobOptions options,
    ILogger<InMemoryBackgroundJobWorker> logger) : BackgroundService
 {
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
+        var workers = Enum.GetValues<BackgroundJobKind>()
+            .SelectMany(kind => Enumerable.Range(
+                    0,
+                    options.ConsumerConcurrency(kind))
+                .Select(_ => ConsumeAsync(kind, stoppingToken)))
+            .ToArray();
+
        try
        {
-            await foreach (var message in transport.ReadAllAsync(stoppingToken))
-            {
-                var result = await runner.RunAsync(message, stoppingToken);
-                if (result.Outcome != BackgroundJobRunOutcome.Retry)
-                    continue;
-
-                await Task.Delay(result.RetryAfter, stoppingToken);
-                await transport.PublishAsync(message, stoppingToken);
-            }
+            await Task.WhenAll(workers);
        }
        catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
        {
            logger.LogInformation("In-memory background job worker is stopping.");
        }
    }
+
+    private async Task ConsumeAsync(
+        BackgroundJobKind kind,
+        CancellationToken cancellationToken)
+    {
+        await foreach (var message in transport.ReadAllAsync(
+                           kind,
+                           cancellationToken))
+        {
+            var result = await runner.RunAsync(message, cancellationToken);
+            if (result.Outcome != BackgroundJobRunOutcome.Retry)
+                continue;
+
+            await Task.Delay(result.RetryAfter, cancellationToken);
+            await transport.PublishAsync(message, cancellationToken);
+        }
+    }
 }
@@ -82,6 +82,7 @@ public sealed class RabbitMqBackgroundJobTransport(
        _connection = await RabbitMqBackgroundJobTopology.CreateConnectionAsync(
            rabbitOptions,
            "jiaowu-background-job-publisher",
+            consumerDispatchConcurrency: 1,
            cancellationToken);
        _channel = await _connection.CreateChannelAsync(
            new CreateChannelOptions(
@@ -153,101 +154,114 @@ public sealed class RabbitMqBackgroundJobWorker(
            var channels = new List<IChannel>();
            try
            {
+                var consumerCount = RabbitMqBackgroundJobTopology.JobKinds.Sum(
+                    jobOptions.ConsumerConcurrency);
                connection = await RabbitMqBackgroundJobTopology.CreateConnectionAsync(
                    rabbitOptions,
                    "jiaowu-background-job-worker",
+                    (ushort)consumerCount,
                    stoppingToken);

                foreach (var kind in RabbitMqBackgroundJobTopology.JobKinds)
                {
-                    var channel = await connection.CreateChannelAsync(
-                        new CreateChannelOptions(
-                            publisherConfirmationsEnabled: false,
-                            publisherConfirmationTrackingEnabled: false),
-                        stoppingToken);
-                    channels.Add(channel);
-                    await RabbitMqBackgroundJobTopology.DeclareAsync(
-                        channel,
-                        jobOptions,
-                        stoppingToken);
-                    await channel.BasicQosAsync(
-                        0,
-                        jobOptions.PrefetchCount,
-                        global: false,
-                        stoppingToken);
-
-                    var consumer = new AsyncEventingBasicConsumer(channel);
-                    consumer.ReceivedAsync += async (_, eventArgs) =>
+                    for (var workerIndex = 0;
+                         workerIndex < jobOptions.ConsumerConcurrency(kind);
+                         workerIndex++)
                    {
-                        using var deliveryCancellation =
-                            CancellationTokenSource.CreateLinkedTokenSource(
-                                eventArgs.CancellationToken,
-                                stoppingToken);
-                        var deliveryToken = deliveryCancellation.Token;
-                        try
+                        var channel = await connection.CreateChannelAsync(
+                            new CreateChannelOptions(
+                                publisherConfirmationsEnabled: false,
+                                publisherConfirmationTrackingEnabled: false),
+                            stoppingToken);
+                        channels.Add(channel);
+                        await RabbitMqBackgroundJobTopology.DeclareAsync(
+                            channel,
+                            jobOptions,
+                            stoppingToken);
+                        await channel.BasicQosAsync(
+                            0,
+                            jobOptions.PrefetchCount,
+                            global: false,
+                            stoppingToken);
+
+                        var consumer = new AsyncEventingBasicConsumer(channel);
+                        consumer.ReceivedAsync += async (_, eventArgs) =>
                        {
-                            var message = JsonSerializer.Deserialize<BackgroundJobEnvelope>(
-                                eventArgs.Body.Span);
-                            if (message is null || message.JobKind != kind)
+                            using var deliveryCancellation =
+                                CancellationTokenSource.CreateLinkedTokenSource(
+                                    eventArgs.CancellationToken,
+                                    stoppingToken);
+                            var deliveryToken = deliveryCancellation.Token;
+                            try
                            {
+                                var message = JsonSerializer
+                                    .Deserialize<BackgroundJobEnvelope>(
+                                        eventArgs.Body.Span);
+                                if (message is null || message.JobKind != kind)
+                                {
+                                    await channel.BasicNackAsync(
+                                        eventArgs.DeliveryTag,
+                                        multiple: false,
+                                        requeue: false,
+                                        deliveryToken);
+                                    return;
+                                }
+
+                                var result = await runner.RunAsync(
+                                    message,
+                                    deliveryToken);
+                                if (result.Outcome ==
+                                    BackgroundJobRunOutcome.Completed)
+                                {
+                                    await channel.BasicAckAsync(
+                                        eventArgs.DeliveryTag,
+                                        multiple: false,
+                                        deliveryToken);
+                                    return;
+                                }
+
+                                await Task.Delay(
+                                    result.RetryAfter,
+                                    deliveryToken);
                                await channel.BasicNackAsync(
                                    eventArgs.DeliveryTag,
                                    multiple: false,
-                                    requeue: false,
+                                    requeue: true,
                                    deliveryToken);
-                                return;
                            }
-
-                            var result = await runner.RunAsync(
-                                message,
-                                deliveryToken);
-                            if (result.Outcome == BackgroundJobRunOutcome.Completed)
+                            catch (OperationCanceledException)
                            {
-                                await channel.BasicAckAsync(
-                                    eventArgs.DeliveryTag,
-                                    multiple: false,
-                                    deliveryToken);
-                                return;
+                                // Closing the channel requeues unacknowledged deliveries.
                            }
-
-                            await Task.Delay(
-                                result.RetryAfter,
-                                deliveryToken);
-                            await channel.BasicNackAsync(
-                                eventArgs.DeliveryTag,
-                                multiple: false,
-                                requeue: true,
-                                deliveryToken);
-                        }
-                        catch (OperationCanceledException)
-                        {
-                            // Closing the channel requeues unacknowledged deliveries.
-                        }
-                        catch (Exception exception)
-                        {
-                            logger.LogError(
-                                exception,
-                                "RabbitMQ delivery for {JobKind} failed and will be requeued.",
-                                kind);
-                            if (channel.IsOpen)
+                            catch (Exception exception)
                            {
+                                logger.LogError(
+                                    exception,
+                                    "RabbitMQ delivery for {JobKind} failed and will be requeued.",
+                                    kind);
+                                if (!channel.IsOpen)
+                                    return;
+
                                await channel.BasicNackAsync(
                                    eventArgs.DeliveryTag,
                                    multiple: false,
                                    requeue: true,
                                    CancellationToken.None);
                            }
-                        }
-                    };
-                    await channel.BasicConsumeAsync(
-                        RabbitMqBackgroundJobTopology.QueueName(jobOptions, kind),
-                        autoAck: false,
-                        consumer,
-                        stoppingToken);
+                        };
+                        await channel.BasicConsumeAsync(
+                            RabbitMqBackgroundJobTopology.QueueName(
+                                jobOptions,
+                                kind),
+                            autoAck: false,
+                            consumer,
+                            stoppingToken);
+                    }
                }

                logger.LogInformation(
-                    "RabbitMQ background job consumers are connected to {HostName}:{Port}.",
+                    "{ConsumerCount} RabbitMQ background job consumers are connected to {HostName}:{Port}.",
+                    consumerCount,
                    rabbitOptions.HostName,
                    rabbitOptions.Port);
                while (connection.IsOpen && !stoppingToken.IsCancellationRequested)
@@ -308,6 +322,7 @@ internal static class RabbitMqBackgroundJobTopology
    public static async Task<IConnection> CreateConnectionAsync(
        RabbitMqOptions options,
        string clientName,
+        ushort consumerDispatchConcurrency,
        CancellationToken cancellationToken)
    {
        var factory = new ConnectionFactory
@@ -321,7 +336,7 @@ internal static class RabbitMqBackgroundJobTopology
            AutomaticRecoveryEnabled = false,
            TopologyRecoveryEnabled = false,
            RequestedHeartbeat = TimeSpan.FromSeconds(30),
-            ConsumerDispatchConcurrency = 1
+            ConsumerDispatchConcurrency = consumerDispatchConcurrency
        };
        if (options.UseTls)
        {
@@ -1000,6 +1000,7 @@ public sealed class AppDbContext(DbContextOptions<AppDbContext> options)
            entity.Property(x => x.LastError).HasMaxLength(2000);
            entity.HasIndex(x => new { x.JobKind, x.JobId }).IsUnique();
            entity.HasIndex(x => new { x.State, x.CreatedAt });
+            entity.HasIndex(x => new { x.State, x.CompletedAt });
            entity.HasIndex(x => x.LeaseExpiresAt);
        });

@@ -2050,6 +2050,10 @@ public sealed class DevelopmentSqliteMigrator(
        """
        CREATE INDEX "IX_BackgroundJobOutboxMessages_State_CreatedAt"
        ON "BackgroundJobOutboxMessages" ("State", "CreatedAt");
+        """,
+        """
+        CREATE INDEX "IX_BackgroundJobOutboxMessages_State_CompletedAt"
+        ON "BackgroundJobOutboxMessages" ("State", "CompletedAt");
        """
    ];
 }
@@ -3478,6 +3478,8 @@ namespace Jiaowu.Api.Infrastructure.Persistence.Migrations.MySql

                    b.HasIndex("State", "CreatedAt");

+                    b.HasIndex("State", "CompletedAt");
+
                    b.ToTable("BackgroundJobOutboxMessages");
                });

@@ -50,6 +50,11 @@ namespace Jiaowu.Api.Infrastructure.Persistence.Migrations.MySql
                name: "IX_BackgroundJobOutboxMessages_State_CreatedAt",
                table: "BackgroundJobOutboxMessages",
                columns: new[] { "State", "CreatedAt" });
+
+            migrationBuilder.CreateIndex(
+                name: "IX_BackgroundJobOutboxMessages_State_CompletedAt",
+                table: "BackgroundJobOutboxMessages",
+                columns: new[] { "State", "CompletedAt" });
        }

        /// <inheritdoc />
@@ -3475,6 +3475,8 @@ namespace Jiaowu.Api.Infrastructure.Persistence.Migrations.MySql

                    b.HasIndex("State", "CreatedAt");

+                    b.HasIndex("State", "CompletedAt");
+
                    b.ToTable("BackgroundJobOutboxMessages");
                });