MassTransit中TransitionTo无法更新SQLServer的EF Core Saga状态

我在项目中集成MassTransit，采用EF Core持久化Saga管理流程：REST API的start-export端点发布ExportProcessInitializeEvent后，RabbitMQ的消费者能依次处理事件并发布下一个事件，事件链执行正常，但SQLServer中的Saga状态记录始终停留在初始状态，后续的TransitionTo操作未更新状态。

核心原因 #

你的事件流转逻辑是通过独立消费者处理并发布下一个事件，而非由Saga状态机直接接收事件并触发状态流转。Saga状态机根本没收到后续事件，自然不会更新数据库中的状态。

修复步骤 #

1. 移除独立消费者配置 #

删除REST API MassTransit配置中所有的AddConsumer和AddRequestClient，避免独立消费者抢先消费事件导致Saga收不到消息：

public static void ConfigureMassTransit(this IServiceCollection services, IConfiguration configuration)
{
    var messageBrokerQueueSettings = configuration.GetSection("MessageBroker:QueueSettings").Get<MessageBrokerQueueSettings>();

    services.AddMassTransit(x =>
    {
        x.UsingRabbitMq((context, cfg) =>
        {
            cfg.Host(messageBrokerQueueSettings.HostName, messageBrokerQueueSettings.VirtualHost, h => {
                h.Username(messageBrokerQueueSettings.UserName);
                h.Password(messageBrokerQueueSettings.Password);
            });

            cfg.ConfigureEndpoints(context);
        });
    });
}

2. 将事件发布逻辑移到Saga状态机中 #

把原本在独立消费者里的发布逻辑，整合到Saga状态机的状态流转动作中，让状态机处理事件后自动触发下一个事件：

#region Flow

During(Initial,
    When(ExportProcessInitializeEvent)
        .Then(x => x.Saga.ExportStartDate = DateTime.UtcNow)
        .TransitionTo(ExportProcessInitializedState)
        .Then(context => context.Publish<DownloadMHTMLsEvent>(new { context.Message.ExportId })));

During(ExportProcessInitializedState,
    When(DownloadMHTMLsEvent)
        .TransitionTo(DownloadMHTMLsState)
        .Then(context => context.Publish<TakeScreenshotsEvent>(new { context.Message.ExportId })));

During(DownloadMHTMLsState,
    When(TakeScreenshotsEvent)
        .TransitionTo(TakeScreenshotsState)
        .Then(context => context.Publish<CreatePPTsEvent>(new { context.Message.ExportId })));

During(TakeScreenshotsState,
    When(CreatePPTsEvent)
        .TransitionTo(CreatePPTsState)
        .Then(context => context.Publish<CreateResultZIPFileEvent>(new { context.Message.ExportId })));

During(CreatePPTsState,
    When(CreateResultZIPFileEvent)
        .TransitionTo(CreateResultZIPFileState)
        .Then(context => context.Publish<NotifyWebsiteInstanceEvent>(new { context.Message.ExportId })));

During(CreateResultZIPFileState,
    When(NotifyWebsiteInstanceEvent)
        .TransitionTo(NotifyWebsiteInstanceState)
        .Then(context => context.Publish<ExportProcessFinalizedEvent>(new { context.Message.ExportId })));

#endregion

3. 为所有事件配置Correlation关联 #

确保Saga能通过ExportId正确关联到对应的实例，在状态机的Event定义部分补充：

Event(() => DownloadMHTMLsEvent, x => x.CorrelateById(context => context.Message.ExportId));
Event(() => TakeScreenshotsEvent, x => x.CorrelateById(context => context.Message.ExportId));
Event(() => CreatePPTsEvent, x => x.CorrelateById(context => context.Message.ExportId));
Event(() => CreateResultZIPFileEvent, x => x.CorrelateById(context => context.Message.ExportId));
Event(() => NotifyWebsiteInstanceEvent, x => x.CorrelateById(context => context.Message.ExportId));
Event(() => ExportProcessFinalizedEvent, x => x.CorrelateById(context => context.Message.ExportId));

4. 删除独立消费者类 #

直接删除所有ExportProcessInitializeEventConsumer、DownloadMHTMLSEventConsumer等独立消费者，相关逻辑已整合到Saga状态机中。

额外检查项 #

• 确认ExportId是Guid类型，与Saga的CorrelationId类型一致，避免关联失败
• 检查EF Core迁移是否正确执行，确保ExportState表包含CorrelationId、CurrentState等必要字段
• 启用MassTransit日志，排查是否存在事件接收失败、Correlation匹配错误等问题

内容的提问来源于stack exchange，提问作者therealsowho