Http POST 上的 MassTransit 托管服务 Saga 状态机超时

Question

您好，通过我的 ProductHttpAggregator 服务控制器提交模型时出现此错误：

抛出异常：System.Private.CoreLib.dll 中的“MassTransit.RequestTimeoutException” 失败：Microsoft.AspNetCore.Diagnostics.DeveloperExceptionPageMiddleware[1] 执行请求时发生未处理的异常。 MassTransit.RequestTimeoutException：等待响应超时，RequestId：00040000-ac13-0242-e33f-08d9ac27912 在 MassTransit.Clients.ResponseHandlerConnectHandle`1.GetTask() 在 ProductHttpAggregator.Controllers.OrderController.Post(OrderViewModel model) 在 C:\Users\tom.fanara\source\repos\SlideXTestSaga\ProductHttpAggregator\Controllers\OrderController.cs:Microsoft.AspNetCore.Mvc.Infrastructure.ActionMethodExecutor.TaskOfIActionResultExecutor 的第 57 行.Execute(IActionResultTypeMapper 映射器，ObjectMethodExecutor 执行器，对象控制器，Object[] 参数)

这是 JSON 请求正文

{
"Id" : "35e12ba0-feb2-42ed-bdd5-322fc507d5f2",
"Timestamp" : "10/15/2021",
"CustomerNumber" : "45678",
"PaymentCardNumber" : "9101112",
"Notes" : "test3"
}

这是控制器中的代码部分。我的 MassTransit 和服务都连接到 RabbitMQ。

 public async Task<IActionResult> Post(OrderViewModel model)
    {
        if (!ModelState.IsValid)
            return BadRequest(ModelState);

        var (accepted, rejected) = await _submitOrderRequestClient.GetResponse<OrderSubmissionAccepted, OrderSubmissionRejected>(new
        {
            OrderId = model.Id,
            InVar.Timestamp,
            model.CustomerNumber,
            model.PaymentCardNumber,
            model.Notes
        });

        if (accepted.IsCompletedSuccessfully)
        {
            var response = await accepted;

            return Accepted(response);
        }
 

这是托管服务设置

public static IHostBuilder CreateHostBuilder(string[] args) =>
        Host.CreateDefaultBuilder(args)
            .ConfigureServices((hostContext, services) =>
            {
                services.AddMassTransit(x =>
                {
                    x.AddDelayedMessageScheduler();

                    x.SetKebabCaseEndpointNameFormatter();

                    // By default, sagas are in-memory, but should be changed to a durable
                    // saga repository.
                    x.SetInMemorySagaRepositoryProvider();

                    x.AddSagaStateMachine(typeof(OrderStateMachine), typeof(OrderStateMachineDefinition));
                    
                    x.UsingRabbitMq((context, cfg) =>
                    {
                        if (IsRunningInContainer)
                            cfg.Host("rabbitmq", h =>
                            {
                                h.Username("guest");
                                h.Password("guest");
                            });

                    cfg.UseDelayedMessageScheduler();

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

               
                services.AddMassTransitHostedService(true); 

            }).ConfigureLogging((hostingContext, logging) =>
            {
                logging.AddSerilog(dispose: true);
                logging.AddConfiguration(hostingContext.Configuration.GetSection("Logging"));
            });

    
}

这里是发布者启动

public void ConfigureServices(IServiceCollection services)
{
    services.AddHealthChecks();

    services.AddApplicationInsightsTelemetry();

    services.ConfigureTelemetryModule<DependencyTrackingTelemetryModule>((module, o) =>
    {
        module.IncludeDiagnosticSourceActivities.Add("MassTransit");
    });

    services.TryAddSingleton(KebabCaseEndpointNameFormatter.Instance);
    services.AddMassTransit(mt =>
    {
        mt.UsingRabbitMq((context, cfg) =>
        {
            cfg.Host(IsRunningInContainer ? "rabbitmq" : "localhost", "/", h =>
                {
                    h.Username("guest");
                    h.Password("guest");
                });
            
            MessageDataDefaults.ExtraTimeToLive = TimeSpan.FromDays(1);
            MessageDataDefaults.Threshold = 2000;
            MessageDataDefaults.AlwaysWriteToRepository = false;

            cfg.UseMessageData(new MongoDbMessageDataRepository(IsRunningInContainer ? "mongodb://mongo" : "mongodb://mongodb:27017", "attachments"));
        });

        mt.AddRequestClient<SubmitOrder>(new Uri($"queue:{KebabCaseEndpointNameFormatter.Instance.Consumer<SubmitOrderConsumer>()}"));

        mt.AddRequestClient<CheckOrder>();
    });

    services.Configure<HealthCheckPublisherOptions>(options =>
    {
        options.Delay = TimeSpan.FromSeconds(2);
        options.Predicate = check => check.Tags.Contains("ready");
    });

    services.AddMassTransitHostedService();

    services.AddOpenApiDocument(cfg => cfg.PostProcess = d => d.Info.Title = "Sample API Site");

    services.AddControllers();
}




enter code here

Answer 1

如果您没有收到回复，可能是您没有启动巴士。配置 MassTransit 时，您应该：

services.AddMassTransitHostedService();

在AddMassTransit(...) 配置之后。