"频道关闭:连接错误"

Question

我在AWS上设置了一个带有两个节点的RabbitMQ集群,并按此处所述启用了HA .然后,我将Elastic Load Balancer映射5672到实例的5672端口,并对实例的15672端口(HTTP管理端口)进行定期运行状况检查.然后,我启动了两个监听器(每个节点一个),每个监听器有4个消费者,并指向spring.rabbitmq.host负载均衡器的DNS.但是,我定期收到以下错误:

[E] [2015-03-14 23:13:23,890] [pool-4-thread-9         ] [CachingConnectionFactory        ] [                                    ] Channel shutdown: connection error
[E] [2015-03-14 23:13:23,891] [pool-4-thread-9         ] [CachingConnectionFactory        ] [                                    ] Channel shutdown: connection error
[E] [2015-03-14 23:13:23,891] [pool-4-thread-9         ] [CachingConnectionFactory        ] [                                    ] Channel shutdown: connection error
[E] [2015-03-14 23:13:23,891] [pool-4-thread-9         ] [CachingConnectionFactory        ] [                                    ] Channel shutdown: connection error
[W] [2015-03-14 23:13:24,758] [impleAsyncTaskExecutor-5] [SimpleMessageListenerContainer  ] [                                    ] Consumer raised exception, processing can restart if the connection factory supports it
com.rabbitmq.client.ShutdownSignalException: connection error
    at com.rabbitmq.client.impl.AMQConnection.startShutdown(AMQConnection.java:717) ~[amqp-client-3.4.2.jar!/:?]
    at com.rabbitmq.client.impl.AMQConnection.shutdown(AMQConnection.java:707) ~[amqp-client-3.4.2.jar!/:?]
    at com.rabbitmq.client.impl.AMQConnection$MainLoop.run(AMQConnection.java:565) ~[amqp-client-3.4.2.jar!/:?]
    at java.lang.Thread.run(Thread.java:745) [?:1.8.0_40]
Caused by: java.io.EOFException
    at java.io.DataInputStream.readUnsignedByte(DataInputStream.java:290) ~[?:1.8.0_40]
    at com.rabbitmq.client.impl.Frame.readFrom(Frame.java:95) ~[amqp-client-3.4.2.jar!/:?]
    at com.rabbitmq.client.impl.SocketFrameHandler.readFrame(SocketFrameHandler.java:139) ~[amqp-client-3.4.2.jar!/:?]
    at com.rabbitmq.client.impl.AMQConnection$MainLoop.run(AMQConnection.java:534) ~[amqp-client-3.4.2.jar!/:?]
    ... 1 more
[I] [2015-03-14 23:13:24,761] [impleAsyncTaskExecutor-5] [SimpleMessageListenerContainer  ] [                                    ] Restarting Consumer: tags=[[amq.ctag-H-1ed6xO3GL7qW58bkLUZA]], channel=Cached Rabbit Channel: AMQChannel(amqp://qube@10.100.43.76:5672qube,1), acknowledgeMode=AUTO local queue size=0
[W] [2015-03-14 23:13:24,762] [impleAsyncTaskExecutor-8] [SimpleMessageListenerContainer  ] [                                    ] Consumer raised exception, processing can restart if the connection factory supports it
com.rabbitmq.client.ShutdownSignalException: connection error
    at com.rabbitmq.client.impl.AMQConnection.startShutdown(AMQConnection.java:717) ~[amqp-client-3.4.2.jar!/:?]
    at com.rabbitmq.client.impl.AMQConnection.shutdown(AMQConnection.java:707) ~[amqp-client-3.4.2.jar!/:?]
    at com.rabbitmq.client.impl.AMQConnection$MainLoop.run(AMQConnection.java:565) ~[amqp-client-3.4.2.jar!/:?]
    at java.lang.Thread.run(Thread.java:745) [?:1.8.0_40]
Caused by: java.io.EOFException
    at java.io.DataInputStream.readUnsignedByte(DataInputStream.java:290) ~[?:1.8.0_40]
    at com.rabbitmq.client.impl.Frame.readFrom(Frame.java:95) ~[amqp-client-3.4.2.jar!/:?]
    at com.rabbitmq.client.impl.SocketFrameHandler.readFrame(SocketFrameHandler.java:139) ~[amqp-client-3.4.2.jar!/:?]
    at com.rabbitmq.client.impl.AMQConnection$MainLoop.run(AMQConnection.java:534) ~[amqp-client-3.4.2.jar!/:?]
    ... 1 more
[I] [2015-03-14 23:13:24,762] [impleAsyncTaskExecutor-8] [SimpleMessageListenerContainer  ] [                                    ] Restarting Consumer: tags=[[amq.ctag-nJAmieDx-kSuxl9arsXiww]], channel=Cached Rabbit Channel: AMQChannel(amqp://qube@10.100.43.76:5672qube,2), acknowledgeMode=AUTO local queue size=0
[W] [2015-03-14 23:13:24,767] [impleAsyncTaskExecutor-7] [SimpleMessageListenerContainer  ] [                                    ] Consumer raised exception, processing can restart if the connection factory supports it
com.rabbitmq.client.ShutdownSignalException: connection error
    at com.rabbitmq.client.impl.AMQConnection.startShutdown(AMQConnection.java:717) ~[amqp-client-3.4.2.jar!/:?]
    at com.rabbitmq.client.impl.AMQConnection.shutdown(AMQConnection.java:707) ~[amqp-client-3.4.2.jar!/:?]
    at com.rabbitmq.client.impl.AMQConnection$MainLoop.run(AMQConnection.java:565) ~[amqp-client-3.4.2.jar!/:?]
    at java.lang.Thread.run(Thread.java:745) [?:1.8.0_40]
Caused by: java.io.EOFException
    at java.io.DataInputStream.readUnsignedByte(DataInputStream.java:290) ~[?:1.8.0_40]
    at com.rabbitmq.client.impl.Frame.readFrom(Frame.java:95) ~[amqp-client-3.4.2.jar!/:?]
    at com.rabbitmq.client.impl.SocketFrameHandler.readFrame(SocketFrameHandler.java:139) ~[amqp-client-3.4.2.jar!/:?]
    at com.rabbitmq.client.impl.AMQConnection$MainLoop.run(AMQConnection.java:534) ~[amqp-client-3.4.2.jar!/:?]
    ... 1 more
[I] [2015-03-14 23:13:24,768] [impleAsyncTaskExecutor-7] [SimpleMessageListenerContainer  ] [                                    ] Restarting Consumer: tags=[[amq.ctag-PAQvN8P57_9oiElyqYRJWw]], channel=Cached Rabbit Channel: AMQChannel(amqp://qube@10.100.43.76:5672qube,3), acknowledgeMode=AUTO local queue size=0
[W] [2015-03-14 23:13:24,768] [impleAsyncTaskExecutor-6] [SimpleMessageListenerContainer  ] [                                    ] Consumer raised exception, processing can restart if the connection factory supports it
com.rabbitmq.client.ShutdownSignalException: connection error
    at com.rabbitmq.client.impl.AMQConnection.startShutdown(AMQConnection.java:717) ~[amqp-client-3.4.2.jar!/:?]
    at com.rabbitmq.client.impl.AMQConnection.shutdown(AMQConnection.java:707) ~[amqp-client-3.4.2.jar!/:?]
    at com.rabbitmq.client.impl.AMQConnection$MainLoop.run(AMQConnection.java:565) ~[amqp-client-3.4.2.jar!/:?]
    at java.lang.Thread.run(Thread.java:745) [?:1.8.0_40]
Caused by: java.io.EOFException
    at java.io.DataInputStream.readUnsignedByte(DataInputStream.java:290) ~[?:1.8.0_40]
    at com.rabbitmq.client.impl.Frame.readFrom(Frame.java:95) ~[amqp-client-3.4.2.jar!/:?]
    at com.rabbitmq.client.impl.SocketFrameHandler.readFrame(SocketFrameHandler.java:139) ~[amqp-client-3.4.2.jar!/:?]
    at com.rabbitmq.client.impl.AMQConnection$MainLoop.run(AMQConnection.java:534) ~[amqp-client-3.4.2.jar!/:?]
    ... 1 more
[I] [2015-03-14 23:13:24,769] [impleAsyncTaskExecutor-6] [SimpleMessageListenerContainer  ] [                                    ] Restarting Consumer: tags=[[amq.ctag-w6apIlL78ViAnTOzx2Qejg]], channel=Cached Rabbit Channel: AMQChannel(amqp://qube@10.100.43.76:5672qube,4), acknowledgeMode=AUTO local queue size=0

如果我将每个消费者直接指向一个节点(不使用负载均衡器),则不会发生这种情况.可能导致这种行为的原因是什么？我怎么能绕过这个呢？

Answer 1

Ping管理端口不会使amqp连接保持活动状态.

您需要将心跳配置为足够低的值,以防止负载均衡器丢弃"空闲"连接.

看起来您可能正在使用Spring Boot(基于属性名称),并且它当前不requestedHeartbeats作为属性公开; 您需要通过将自己的rabbitConnectionFactorybean与基础rabbitmq连接工厂心跳设置连接来覆盖引导工厂.有关详细信息,请参阅文档

也就是说,使用带有Spring AMQP的负载均衡器没有任何好处,因为它使用单个长期连接.考虑在addresses属性中配置两个实例,如果第一个实例不可用,它将故障转移到第二个实例.