使用 RabbitMq 的 MassTransit 在发生错误时未重试

Question

我在 MassTransit 上使用 rabbitmq 时遇到了重试问题。

我已经拿到了代码配置：

        private static void ConfigureMassTransitWithRabbitMq(this IServiceCollection services, string host, string username, string pwd, string cluster, Assembly assembly)
        {   
            services.AddMassTransit(config =>
            {
                config.AddConsumers(assembly);

                config.SetKebabCaseEndpointNameFormatter();

                config.UsingRabbitMq((context, rabbitMqConfig) =>
                {   
                    rabbitMqConfig.UseRetry(e => Retry.Immediate(3));
                    rabbitMqConfig.ConfigureEndpoints(context);
                    rabbitMqConfig.ConfigurePublish(pipe => 
                    { 
                        pipe.UseExecute(context1 => 
                        { 
                            if (context1.CorrelationId == null) 
                            { 
                                context1.CorrelationId = Guid.NewGuid(); 
                            } 
                        }); 
                    });

                    rabbitMqConfig.Host(new Uri(host), hst =>
                    {
                        hst.Username(username);
                        hst.Password(pwd);
                        hst.PublisherConfirmation = true;
                        hst.Heartbeat(5);
                        hst.UseCluster(c =>
                        {
                            var clusters = cluster?.Split(';');

                            if (cluster == null || clusters.Length == 0)
                                return;

                            foreach (var item in clusters)
                                c.Node(item);
                        });
                    });

                    rabbitMqConfig.ManagementEndpoint((conf) =>
                    {
                        conf.ConfigurePublish((p) =>
                        {
                            p.UseRetry((r) =>
                            {
                                r.Exponential(3, TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(15), TimeSpan.FromSeconds(5));
                            });
                        });
                    });
                });
            });

            services.AddMassTransitHostedService();
        }

我通过组装添加我的消费者。

我只是尝试使用：

• rabbitMqConfig.UseRetry(e => Retry.Immediate(3)); 当我配置 MassTransit 时

但在这种情况下，de 配置保持无限循环

---

我尝试使用这个：

        try
        {
            var order = _mapper.Map<object>(context.Message);
            await Execute(order);
            await context.ConsumeCompleted;
        }
        catch (Exception)
        {
            var retry = context.GetRetryCount();
           
            if (maxAttempts > 3)
            {
                throw;
            }

            await context.Redeliver(TimeSpan.FromSeconds(5));
        }

我把这段代码放在 MassTransitConfiguration 中：

rabbitMqConfig.UseDelayedExchangeMessageScheduler();

---

在最后一种情况下，我得到了重试，但我从来没有得到重试的数量，所以我无法控制它。我也尝试通过标题和其他属性来获得这个数量，但我总是得到 0。

谁能帮帮我？！

Answer 1

您的配置顺序错误，根据您上面的示例更正的配置如下所示：

private static void ConfigureMassTransitWithRabbitMq(this IServiceCollection services, string host, string username, string pwd, string cluster, Assembly assembly)
{   
    services.AddMassTransit(config =>
    {
        config.AddConsumers(assembly);

        config.SetKebabCaseEndpointNameFormatter();

        config.UsingRabbitMq((context, rabbitMqConfig) =>
        {   
            rabbitMqConfig.Host(new Uri(host), hst =>
            {
                hst.Username(username);
                hst.Password(pwd);
                hst.Heartbeat(5);
                hst.UseCluster(c =>
                {
                    var clusters = cluster?.Split(';');

                    if (cluster == null || clusters.Length == 0)
                        return;

                    foreach (var item in clusters)
                        c.Node(item);
                });
            });

            rabbitMqConfig.ConfigurePublish(pipe => 
            { 
                pipe.UseExecute(context1 => 
                { 
                    context1.CorrelationId ??= Guid.NewGuid(); 
                }); 
            });

            rabbitMqConfig.UseRetry(e => Retry.Immediate(3));

            rabbitMqConfig.ConfigureEndpoints(context);
        });
    });
    services.AddMassTransitHostedService();
}

我还删除了任何不需要的东西，要么是默认行为，要么是未使用。