• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    公众号
登陆 注册

.net/c#RabbitMQ连接断开处理-断线重连（转载）

原作者: [db:作者] 来自: [db:来源] 收藏 邀请

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

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

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

************** 异常文本 **************
RabbitMQ.Client.Exceptions.AlreadyClosedException: Already closed: The AMQP operation was interrupted: AMQP close-reason, initiated by Peer, code=320, text="CONNECTION_FORCED - broker forced connection closure with reason 'shutdown'", classId=0, methodId=0, cause=
   在 RabbitMQ.Client.Impl.SessionBase.Transmit(Command cmd)
   在 RabbitMQ.Client.Impl.ModelBase.ModelSend(MethodBase method, ContentHeaderBase header, Byte[] body)
   在 RabbitMQ.Client.Impl.ModelBase.BasicPublish(String exchange, String routingKey, Boolean mandatory, IBasicProperties basicProperties, Byte[] body)
   在 RabbitMQ.Client.Impl.ModelBase.BasicPublish(String exchange, String routingKey, IBasicProperties basicProperties, Byte[] body)
   在 rabbitMQ_Publish.Form1.button1_Click(Object sender, EventArgs e) 位置 C:\project\my\RabbitMQ-demo\rabbitMQ-Publish\Form1.cs:行号 37
   在 System.Windows.Forms.Control.OnClick(EventArgs e)
   在 System.Windows.Forms.Button.OnClick(EventArgs e)
   在 System.Windows.Forms.Button.PerformClick()
   在 System.Windows.Forms.Form.ProcessDialogKey(Keys keyData)
   在 System.Windows.Forms.TextBoxBase.ProcessDialogKey(Keys keyData)
   在 System.Windows.Forms.Control.PreProcessMessage(Message& msg)
   在 System.Windows.Forms.Control.PreProcessControlMessageInternal(Control target, Message& msg)
   在 System.Windows.Forms.Application.ThreadContext.PreTranslateMessage(MSG& msg)






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

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());

  5.             AutorecoveringConnection self = this;
  6.             EventHandler<ShutdownEventArgs> recoveryListener = (_, args) =>
  7.             {
  8.                 lock (recoveryLockTarget)
  9.                 {
  10.                     if (ShouldTriggerConnectionRecovery(args))
  11.                     {
  12.                         try
  13.                         {
  14.                             self.BeginAutomaticRecovery();
  15.                         }
  16.                         catch (Exception e)
  17.                         {
  18.                             // TODO: logging
  19.                             Console.WriteLine("BeginAutomaticRecovery() failed: {0}", e);
  20.                         }
  21.                     }
  22.                 }
  23.             };
  24.             lock (m_eventLock)
  25.             {
  26.                 ConnectionShutdown += recoveryListener;
  27.                 if (!m_recordedShutdownEventHandlers.Contains(recoveryListener))
  28.                 {
  29.                     m_recordedShutdownEventHandlers.Add(recoveryListener);
  30.                 }
  31.             }
  32.         }
复制代码

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

  1. protected void PerformAutomaticRecovery()
  2.         {
  3.             lock (recoveryLockTarget)
  4.             {
  5.                 RecoverConnectionDelegate();
  6.                 RecoverConnectionShutdownHandlers();
  7.                 RecoverConnectionBlockedHandlers();
  8.                 RecoverConnectionUnblockedHandlers();

  10.                 RecoverModels();
  11.                 if (m_factory.TopologyRecoveryEnabled)
  12.                 {
  13.                     RecoverEntities();
  14.                     RecoverConsumers();
  15.                 }

  17.                 RunRecoveryEventHandlers();
  18.             }
  19.         }
复制代码



这个方法中调用的是

  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

鲜花
握手
雷人
路过
鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
More+
上一篇：
DDKbuild的时候C2220警告当错误处理发布时间：2022-07-14
下一篇：
C#打印代码运行时间发布时间：2022-07-14
热门推荐
More+
热门话题
More+
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服（服务时间 9:00～18:00）

在线QQ客服
地址：深圳市南山区西丽大学城创智工业园
电邮：jeky_zhao#qq.com
移动电话：139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap