Rabbitmq 官方给的NET consumer示例代码如下,但使用过程,会遇到connection断开的问题,一旦断开,这个代码就会报错,就会导致消费者或者生产者挂掉。

下图是生产者发送消息,我手动停止了rabbitmq,然后又重新启动了rabbitmq,大概等启动成功以后,为了防止服务没有完全启动,我又等待了10秒钟

服务完全启动成功以后,我尝试重新发送一些消息,报错,如下:

.net/c# RabbitMQ 连接断开处理-断线重连(转载)-LMLPHP

那么如何会异常恢复呢?或者说断线重连呢?

RabbitMQ NET Client的源码,研究发现一种自动的错误恢复机制 AutomaticRecoveryEnabled = true 使用方式如下

  1. var factory = new ConnectionFactory() { HostName = "localhost", AutomaticRecoveryEnabled = true };

复制代码

具体的恢复机制如下

1.在AutoRecoveringConnection初始化时,在链接关闭事件委托上增加断开处理

  1. public void init()
  2. {
  3. m_delegate = new Connection(m_factory, false, m_factory.CreateFrameHandler());
  4. AutorecoveringConnection self = this;
  5. EventHandler<ShutdownEventArgs> recoveryListener = (_, args) =>
  6. {
  7. lock (recoveryLockTarget)
  8. {
  9. if (ShouldTriggerConnectionRecovery(args))
  10. {
  11. try
  12. {
  13. self.BeginAutomaticRecovery();
  14. }
  15. catch (Exception e)
  16. {
  17. // TODO: logging
  18. Console.WriteLine("BeginAutomaticRecovery() failed: {0}", e);
  19. }
  20. }
  21. }
  22. };
  23. lock (m_eventLock)
  24. {
  25. ConnectionShutdown += recoveryListener;
  26. if (!m_recordedShutdownEventHandlers.Contains(recoveryListener))
  27. {
  28. m_recordedShutdownEventHandlers.Add(recoveryListener);
  29. }
  30. }
  31. }

复制代码

观察调用的方式BeginAutomaticRecovery,可以看到这个方法内部调用了PerformAutomaticRecovery方法。我们直接看这个方法的内容,其中第一个调用的是方法RecoverConnectionDelegate

  1. protected void PerformAutomaticRecovery()
  2. {
  3. lock (recoveryLockTarget)
  4. {
  5. RecoverConnectionDelegate();
  6. RecoverConnectionShutdownHandlers();
  7. RecoverConnectionBlockedHandlers();
  8. RecoverConnectionUnblockedHandlers();
  9. RecoverModels();
  10. if (m_factory.TopologyRecoveryEnabled)
  11. {
  12. RecoverEntities();
  13. RecoverConsumers();
  14. }
  15. RunRecoveryEventHandlers();
  16. }
  17. }

复制代码

这个方法中调用的是

  1. protected void RecoverConnectionDelegate()
  2. {
  3. bool recovering = true;
  4. while (recovering)
  5. {
  6. try
  7. {
  8. m_delegate = new Connection(m_factory, false, m_factory.CreateFrameHandler());
  9. recovering = false;
  10. }
  11. catch (Exception)
  12. {
  13. // TODO: exponential back-off
  14. Thread.Sleep(m_factory.NetworkRecoveryInterval);
  15. // TODO: provide a way to handle these exceptions
  16. }
  17. }
  18. }

复制代码

可以看出,它是执行了死循环,直到连接重新打开,当然,如果遇到异常,它会调用Thread.Sleep来等待一下,然后再次执行连接恢复。

转载自: https://www.itsvse.com/thread-4636-1-1.html

05-07 15:17