异步消息与事件驱动（对照 Spring AMQP）

这一课你将能做到

1. 把「提交代码 → 判题」从同步 HTTP 改造为消息驱动的异步链路（削峰 + 解耦）
2. 用 MassTransit 声明消息契约、发布事件、写消费者，并配置重试与死信
3. 说清 at-least-once 语义下幂等消费的实现套路，对照 Spring 的 @RabbitListener 体系

为什么要学

判题是慢操作（编译 + 沙箱执行可能几秒），如果提交接口同步等结果，高峰期线程池瞬间被占满、QPS 崩掉。把「提交」和「判题」拆开：接口只做「落库 + 发消息」立即返回 PENDING，判题 Worker 从队列消费慢慢干——这就是削峰；判题引擎换语言实现、扩缩容，都不动 API 代码——这就是解耦。

Java 侧这套是 RabbitMQ + @RabbitTemplate/@RabbitListener；.NET 侧的主流抽象是 MassTransit（类库级的消息总线，类似 Spring Cloud Stream 的定位），底层接 RabbitMQ/Azure Service Bus/Kafka。

先建立图像

用户提交                  API 实例                 RabbitMQ              判题 Worker
   │ POST /submissions      │                        │                     │
   │ ────────────────────→ 落库(PENDING)               │                     │
   │ ←─ 201 PENDING ────────│ ── publish ─────────→ [judge.queue]           │
   │                        │                        │ ── consume ───────→ 沙箱判题
   │                        │                        │ ←─ ack ──────────── 更新DB
   │ ←══ SignalR 推送 AC ══════════════════════════════════════════════════│

交付语义：RabbitMQ 默认 at-least-once——Worker 处理完才 ack，处理到一半宕机消息会重投。所以消费逻辑必须幂等（同一提交判两次不产生副作用）。

跟着做

步骤 1：定义消息契约

dotnet add package MassTransit.RabbitMQ

// 独立 Contracts 项目，发布方与消费方共享（版本演进靠加字段不靠改字段）
public record SubmissionSubmitted(long SubmissionId, long ProblemId, string Language, string Code);

public record SubmissionJudged(long SubmissionId, string Status, int TimeMs, string UserName);

步骤 2：API 侧发布事件

// Program.cs
builder.Services.AddMassTransit(x =>
{
    x.SetKebabCaseEndpointNameFormatter();
    x.UsingRabbitMq((ctx, cfg) =>
    {
        cfg.Host("localhost", "/", h => { h.Username("guest"); h.Password("guest"); });
        cfg.ConfigureEndpoints(ctx);        // 自动按消费者创建交换机/队列
    });
});

// 提交接口：落库 + publish，双成功才返回 201
api.MapPost("/", async (SubmitDto dto, OjDbContext db, IPublishBus bus) =>
{
    var submission = new Submission { ProblemId = dto.ProblemId, Language = dto.Language, Status = "PENDING" };
    db.Submissions.Add(submission);
    await db.SaveChangesAsync();

    await bus.Publish(new SubmissionSubmitted(submission.Id, dto.ProblemId, dto.Language, dto.Code));
    return Results.Created($"/api/submissions/{submission.Id}", new { submission.Id, submission.Status });
});

步骤 3：Worker 侧消费（BackgroundService + Consumer 二选一）

public class SubmissionSubmittedConsumer(OjDbContext db, IPublishBus bus) : IConsumer<SubmissionSubmitted>
{
    public async Task Consume(ConsumeContext<SubmissionSubmitted> ctx)
    {
        var msg = ctx.Message;

        // 幂等防线：状态已是终态就直接 ack（消息重投时的第二道保险）
        var submission = await db.Submissions.FirstOrDefaultAsync(s => s.Id == msg.SubmissionId);
        if (submission is null || submission.Status is not ("PENDING" or "QUEUED")) return;

        var result = await SandboxRunner.RunAsync(msg.Code, msg.Language, ctx.CancellationToken);

        submission.Status = result.Status;      // 实体被跟踪，SaveChanges 生成 UPDATE
        submission.TimeMs = result.TimeMs;
        await db.SaveChangesAsync();

        await bus.Publish(new SubmissionJudged(msg.SubmissionId, result.Status,
            result.TimeMs, submission.UserName));  // 下游: SignalR 推送/排行榜/统计各自订阅
    }
}

// Program.cs（Worker 项目）
builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<SubmissionSubmittedConsumer>();
    x.UsingRabbitMq((ctx, cfg) =>
    {
        cfg.UseMessageRetry(r => r.Exponential(5, TimeSpan.FromSeconds(1),
            TimeSpan.FromMinutes(2), TimeSpan.FromSeconds(5)));   // 指数退避重试 5 次
        cfg.UseDelayedRedelivery(r => r.Intervals(TimeSpan.FromMinutes(1),
            TimeSpan.FromMinutes(5), TimeSpan.FromMinutes(15)));  // 更长周期靠延迟重投
        cfg.ConfigureEndpoints(ctx);
    });
});

Spring 对照：IConsumer<T> ≙ @RabbitListener 方法；IPublishBus.Publish ≙ rabbitTemplate.convertAndSend；UseMessageRetry ≙ 容器重试拦截器；重试耗尽后消息进入 _error 队列 ≙ 死信队列（DLX）。

步骤 4：幂等消费的标准写法

重投不可避免，消费侧要能扛住重复：

// 方案 A：状态机式（如上，PENDING 才处理，天然幂等）
// 方案 B：Inbox 表去重（通用）
var processed = await db.InboxMessages.FindAsync(ctx.MessageId);
if (processed is not null) return;                       // 见过就跳过
db.InboxMessages.Add(new InboxMessage(ctx.MessageId, DateTime.UtcNow));
await db.SaveChangesAsync();                             // 消息ID唯一索引兜底并发

步骤 5：确认死信与监控

x.AddConfigureEndpointsCallback((name, cfg) =>
{
    cfg.UseDeadLetterQueue();          // 失败消息进 DLQ，人工/定时补偿
});

运维检查点：RabbitMQ 管理页看 judge.queue 积压深度、消费速率、_error 队列有无堆积。积压持续增长 = 判题能力不足，横向加 Worker 实例即可（队列自动负载均衡）。

讲清楚要点

1. 消息不是数据库，别当存储用：队列只做「暂存与转运」，业务数据以数据库为准；消息体传 ID + 最小字段，消费方回查数据库拿全量（避免大消息与陈旧快照）。大文件（代码、测试数据）传对象存储 URL，不塞消息体。
2. at-least-once + 幂等 = 工程上的 exactly-once：投递语义没法靠 MQ 单方面精确一次，靠「消费幂等」补齐。判题场景幂等点选「状态机 + Inbox 表」双保险。
3. 事件命名与职责：SubmissionSubmitted 是事实（已发生），订阅方决定做什么（入队、统计、通知）——发布方不指定消费方行为，这是事件驱动与命令（Send 点对点）的本质区别。MassTransit 里 Publish 广播事实、Send 定点投递命令，对应 Spring 里 topic 交换机 vs direct 路由的设计决策。
4. 重试要分层：瞬时错误（网络抖动）当场指数退避；持续错误（Bug、数据缺失）重试无意义，快速进 DLQ 保住吞吐；补偿动作（人工修数、重放）基于 DLQ 消息做。

你可能会懵的地方

• 进程内异步（@Async / Task.Run）和 MQ 有什么区别？ 进程内异步共享同一进程的线程池与生命周期，进程重启任务就丢；MQ 跨进程、可持久化、可独立扩缩容。判题必须跨进程（沙箱环境隔离 + 独立伸缩），所以用 MQ；「提交后发欢迎邮件」这类轻任务进程内异步即可。

• 为什么不用 Redis List/Stream 当队列？ 轻量场景可以（第 6 课的 Redis 就在栈内）。但缺 ack 语义、死信、延迟重投这些要自己造；RabbitMQ 消费确认与死信是产品级。选型逻辑与 Spring 项目「Redis Stream 还是 RabbitMQ」一致：吞吐小、容忍丢重试简单 → Redis；要可靠交付 → 专业 MQ。

• 消息顺序怎么保证？ 同一提交的「提交→判题→通知」有顺序依赖时，把该用户的全部消息路由到同一队列分区（RabbitMQ 用 consistent-hash exchange 按 SubmissionId 路由；Kafka 按 key 分区）。代价是并行度受分区数限制，能用状态机幂等解决的就不要引入顺序约束。

• 分布式事务（落库成功但发消息失败）怎么办？ 标准答案是 Outbox 模式：业务数据和「待发消息」在同一个本地事务里落库，后台轮询/变更捕获（Debezium 类）把 outbox 表的消息真正投出去。MassTransit 内置 AddOutbox/EF Outbox 支持。简单场景先接受「发消息失败返回 500 让用户重试」也是务实解。

• MessagePack/Avro 要不要上？ 起步用 JSON（可读、可调试），吞吐或带宽成为瓶颈再换 MessagePack（MassTransit 一行配置）。Schema 演进纪律（只加字段、消费方容忍未知字段）比编码格式重要。

错法对照

错法	问题	正解
消费者处理完不抛错也不 catch，异常直接裸奔	消息反复重投打爆队列	分层：瞬时错重试，确定性错进 DLQ + 告警
把整个提交代码 + 全部用例塞进消息	消息体膨胀、broker 压力大	消息传 ID，消费方查库/对象存储
消费逻辑非幂等（每次都 +1 计数）	重投导致数据重复累加	状态机判断或 Inbox 消息表去重
同步发布消息后仍同步等判题结果	消息化了但没解耦，线程照旧被占	接口立即返回 PENDING，结果走推送
队列不设上限也不监控积压	Worker 挂了没人发现，用户永远 PENDING	积压告警 + 消费速率监控 + PENDING 超时兜底任务

卡点急救

1. 消息发出去了但消费者没收到：核对两端 cfg.Host 与 virtual host；ConfigureEndpoints 生成的队列名格式（kebab-case）与预期不一致时，显式 x.Message<SubmissionSubmitted>(m => m.SetEntityName(...)) 统一。
2. 重试风暴（同一消息疯狂重投）：确认异常被 MassTransit 重试策略包裹而非被自己 try/catch 吞掉——吞掉 = ack，抛出 = 重试，语义要先想清楚。
3. 本地无 RabbitMQ 起不来：docker run -d -p 5672:5672 -p 15672:15672 rabbitmq:3-management；CI 或本地开发可用 MassTransit 的 UsingInMemory 传输替换，零外部依赖跑通链路。

面试常问

• 消息队列三用途与判题平台的具体落点？ 削峰（提交高峰把判题请求排队，Worker 按沙箱容量匀速消费）、解耦（API 不依赖判题实现，事件发布后推送/统计/排行榜各自订阅）、异步（提交接口只承担落库+投递）。配套设计：at-least-once 交付 + 消费幂等 + DLQ 补偿。

• 如何实现幂等消费？ 三层防线：业务状态机（只有 PENDING/QUEUED 状态才执行判题）、消息 ID Inbox 去重表（唯一索引挡并发重投）、最终结果以数据库覆盖写为准（UPDATE 天然幂等）。对应 Java 侧同样的 Inbox/状态机模式，与具体 MQ 无关。

• Outbox 模式解决什么问题？ 「数据库写成功但消息发送失败」的双写不一致。把消息先和业务数据写进同一个本地事务（outbox 表），再由独立进程把 outbox 消息可靠投递到 MQ，用「本地事务 + 轮询重投」替代不可靠的分布式事务。MassTransit 提供 EF Core Outbox 开箱支持，Java 侧对应 Transactional Outbox + Debezium CDC 方案。

自测三问

1. Publish 和 Send 在 MassTransit 里语义差在哪？
2. 消费者抛异常时消息会怎样？被 try/catch 吞掉时呢？
3. 为什么「先更新数据库再发消息」不保险，Outbox 怎么解决？

小结一句话

消息驱动的判题链路 = 落库即返回 + 事件广播 + 幂等消费 + 死信兜底；MassTransit 之于 .NET 约等于 Spring AMQP/Cloud Stream 之于 Java，交付语义与幂等设计两边完全通用。

延伸

• 官方文档：.NET 后台服务与工作者模板
• 下一课：微服务与跨服务通信——gRPC、YARP 网关、弹性重试，对照 Spring Cloud。