使用Mass Transit与RabbitMQ时遇连接被远程主机强制关闭问题

问题描述 #

新装RabbitMQ后，原本使用MassTransit发送消息正常，突然抛出RabbitMQ.Client.Exceptions.OperationInterruptedException异常，核心错误为**"An existing connection was forcibly closed by the remote host"**（连接被远程主机强制关闭）。Windows防火墙已关闭，无法定位问题，请求协助。

异常栈信息 #

RabbitMQ.Client.Exceptions.OperationInterruptedException
HResult=0x80131500   Message=The AMQP operation was interrupted: AMQP close-reason, initiated by Library, code=541, text='Unexpected Exception', classId=0, methodId=0, cause=System.IO.IOException: Unable to read data from the transport connection: An existing connection was forcibly closed by the remote host..  ---> System.Net.Sockets.SocketException (10054): An existing connection was forcibly closed by the remote host.    at System.Net.Sockets.NetworkStream.Read(Byte[] buffer, Int32 offset, Int32 count)    --- End of inner exception stack trace ---    at System.Net.Sockets.NetworkStream.Read(Byte[] buffer, Int32 offset, Int32 count)    at System.IO.BufferedStream.ReadByteSlow()    at RabbitMQ.Client.Impl.InboundFrame.ReadFrom(Stream reader, Byte[] frameHeaderBuffer, ArrayPool`1 pool, UInt32 maxMessageSize)    at RabbitMQ.Client.Impl.SocketFrameHandler.ReadFrame()    at RabbitMQ.Client.Framing.Impl.Connection.MainLoopIteration()    at RabbitMQ.Client.Framing.Impl.Connection.MainLoop()   Source=RabbitMQ.Client   StackTrace:    at RabbitMQ.Client.Impl.SimpleBlockingRpcContinuation.GetReply(TimeSpan timeout)    at RabbitMQ.Client.Impl.ModelBase.ModelRpc(MethodBase method, ContentHeaderBase header, Byte[] body)    at RabbitMQ.Client.Framing.Impl.Model._Private_ChannelOpen(String outOfBand)    at RabbitMQ.Client.Framing.Impl.Connection.CreateModel() at MassTransit.RabbitMqTransport.RabbitMqConnectionContext.<CreateModel>b__26_0() at MassTransit.Util.ChannelExecutor.SynchronousFuture`1.Run()
--- End of stack trace from previous location ---    at MassTransit.Util.ChannelExecutor.d__14`1.MoveNext()    at MassTransit.RabbitMqTransport.RabbitMqConnectionContext.<CreateModel>d__26.MoveNext() at MassTransit.RabbitMqTransport.RabbitMqConnectionContext.<CreateModelContext>d__27.MoveNext() at MassTransit.SupervisorExtensions.CreateAgentPipe`2.d__4.MoveNext()
at MassTransit.Internals.TaskExtensions.<>c__DisplayClass2_0`1.<<OrCanceled>g__WaitAsync|0>d.MoveNext() at MassTransit.SupervisorExtensions.<CreateAgent>d__5`2.MoveNext()
at MassTransit.Internals.TaskExtensions.<>c__DisplayClass2_0`1.<<OrCanceled>g__WaitAsync|0>d.MoveNext() at MassTransit.RabbitMqTransport.ModelContextFactory.<CreateSharedModel>d__4.MoveNext() at MassTransit.Agents.PipeContextSupervisor`1.d__7.MoveNext()
at MassTransit.Agents.PipeContextSupervisor`1.<Send>d__7.MoveNext()    at MassTransit.Agents.PipeContextSupervisor`1.d__7.MoveNext()
at MassTransit.SupervisorExtensions.<>c__DisplayClass5_0`2.<<CreateAgent>g__HandleSupervisorTask|0>d.MoveNext() at MassTransit.SupervisorExtensions.<CreateAgent>d__5`2.MoveNext()
at MassTransit.RabbitMqTransport.ScopeModelContextFactory.d__4.MoveNext()
at MassTransit.Agents.PipeContextSupervisor`1.<Send>d__7.MoveNext()    at MassTransit.Agents.PipeContextSupervisor`1.d__7.MoveNext()
at MassTransit.Agents.PipeContextSupervisor`1.d__7.MoveNext()
at MassTransit.Transports.HostConfigurationRetryExtensions.d__0.MoveNext()

MassTransit配置代码 #

conf.UsingRabbitMq((context, cfg) =>
{
    cfg.Host("localhost", "/", h =>
    {
        h.Username("guest");
        h.Password("guest");
    });
});

RabbitMQ服务状态详情 #

• 节点状态：rabbit@DEV-BEGUEST 运行中
• 基础信息：OS PID 6808，Windows系统，运行时长818秒，非维护模式
• 版本信息：RabbitMQ 3.11.16（官方支持），Erlang/OTP 26 [erts-14.0]，OpenSSL 3.1.0
• 已启用插件：rabbitmq_management、amqp_client、rabbitmq_web_dispatch、cowboy、cowlib、rabbitmq_management_agent
• 文件路径：
  • 数据目录：c:/Users/Xavi/AppData/Roaming/RabbitMQ/db/rabbit@DEV-BEGUEST-mnesia
  • 配置文件：c:/Users/Xavi/AppData/Roaming/RabbitMQ/advanced.config
  • 日志文件：c:/Users/Xavi/AppData/Roaming/RabbitMQ/log/rabbit@DEV-BEGUEST.log、c:/Users/Xavi/AppData/Roaming/RabbitMQ/log/rabbit@DEV-BEGUEST_upgrade.log
• 资源状态：
  • 内存：使用0.0885GB，远低于5.9848GB的高水位线
  • 文件描述符：已用2（上限65439），套接字已用0（上限58893）
  • 磁盘：剩余67.6863GB，远高于0.05GB的低水位线
• 统计信息：连接数0，队列数0，虚拟主机数1
• 监听端口：[::]:15672（HTTP API）、[::]:25672（集群通信）、[::]:5672/0.0.0.0:5672（AMQP协议）

排查与解决步骤 #

1. 修复Erlang与RabbitMQ版本兼容性
   RabbitMQ 3.11.x官方推荐Erlang 25.x版本，当前使用的Erlang 26存在版本兼容问题，会导致连接异常。建议降级Erlang到25.3.2.8（RabbitMQ 3.11.16对应的兼容版本），然后重启RabbitMQ服务。

2. 优化MassTransit连接配置
   添加心跳、超时设置与重试机制，避免连接闲置被主动断开：

   conf.UsingRabbitMq((context, cfg) =>
   {
       cfg.Host("localhost", "/", h =>
       {
           h.Username("guest");
           h.Password("guest");
           h.Heartbeat(30); // 设置心跳为30秒
           h.ConnectionTimeout(TimeSpan.FromSeconds(10)); // 连接超时10秒
       });
       cfg.UseRetry(r => r.Interval(3, TimeSpan.FromSeconds(2))); // 添加3次间隔2秒的重试
   });
   

3. 查看RabbitMQ日志详情
   打开rabbit@DEV-BEGUEST.log，搜索error或connection closed关键词，定位RabbitMQ主动关闭连接的具体原因（如认证失败、协议不匹配等）。

4. 重置RabbitMQ用户权限
   执行以下命令重置guest用户的权限，避免权限异常导致连接失败：

   rabbitmqctl set_permissions -p / guest ".*" ".*" ".*"
   

5. 重启服务与应用
   完全停止RabbitMQ服务（包括Erlang进程），等待10秒后重新启动，同时重启应用程序清除旧连接缓存。

内容的提问来源于stack exchange，提问作者Xavier Casals