RabbitMQ 关闭时的 Masstransit 重新连接策略

Question

我有以下不同的场景：

场景 1

RabbitMQ 已启动。

消费者通过 MassTransit 正确启动和订阅 RabbitMQ。

RabbitMQ 出现故障。

RabbitMQ 又涨了。

如果 Consumer 尝试使用 busControl.StartAsync(); 订阅，那么它将不再重新连接

如果消费者尝试使用 busControl.Start(); 订阅，那么它将正确地重新连接。

已经觉得这有点奇怪.. 但还好。

场景 2

RabbitMQ 已关闭

Consumer 上去尝试用 busControl.Start(); 订阅，会因为 RabbitMQ down 而失败

RabbitMQ 涨了

消费者无法重新连接到 RabbitMQ

但我需要以某种方式涵盖此案..

注册：

private static void RegisterRabbitMQ(SimpleInjectorContainer container)
{
    container.AddMassTransit(x =>
    {
        x.AddBus(() => Bus.Factory.CreateUsingRabbitMq(cfg =>
        {
            var host = cfg.Host(ConfigurationValuesProvider.Current.Get("RabbitMQHostName"), hostConfigurator =>
            {
#if !DEBUG
                hostConfigurator.Username(ConfigurationValuesProvider.Current.Get("RabbitMQUsername"));
                hostConfigurator.Password(ConfigurationValuesProvider.Current.Get("RabbitMQPassword"));

                hostConfigurator.UseCluster(c =>
                {
                    string[] hostnames = ConfigurationValuesProvider.Current.Get("RabbitMQNodes").Split(';');
                    c.ClusterMembers = hostnames;
                });
#endif
            });
        }));
    });
}

出发巴士：

try
{
    RegisterRabbitMQ(container);
    var busControl = container.GetInstance<IBusControl>();
    busControl.Start();
}
catch (Exception ex)
{
    Log.Seq.Error("RabbitMQ broker is not reachable, BusControl cannot be started. ", ex);
}

提前致谢！

Answer 1

我们在生产中使用 Mass Transit 已经有一段时间了。一旦 Mass Transit 连接到 Rabbit，公共交通的连接管理就非常可靠。当您的消费者服务出现时，虚拟机 /Containers 的启动和停止以及兔子宕机是我认为的真正问题。

我们解决此问题的方法是将消费者的启动移动到服务的不同线程并实施重试操作。这将使您的服务成功启动并进入循环并等待兔子出现。重要的是，我们记录了该服务无法连接到 rabbit，因此我们可以看到这一点。

代码

private async Task StartAsync()
{
    await Task.Run(() => {

        bool isConnected = false;
        while(!isConnected)
        {
            try
            {
                ServiceBus = CreateServiceBus();
                ServiceBus?.Start();
                isConnected = true;
                _loggingService.Info("Messaging service bus started.");
            }
            catch(Exception ex)
            {
                _loggingService.Error("Cannot connect to message broker host.", ex);
                Thread.Sleep(Defined_Time);
            }
        }
    });
}

那么配置你的总线就像你在演示中看到的那样

private IBusControl CreateServiceBus()
{
    var url = string.Format(CultureInfo.InvariantCulture, "rabbitmq://{0}", _messagingServerName);
    var uri = new Uri(url);

   _loggingService.DebugFormat("Starting messaging service bus: \"{0}\". Username=\"{1}\".", url, _messagingUsername);

    return Bus.Factory.CreateUsingRabbitMq(sbc =>
    {
        Host = sbc.Host(uri, h =>
        {
            h.Username(_messagingUsername);
            h.Password(_messagingPassword);
        });

        //Configure end points
    });
}

希望对您有所帮助。