«Отключение канала: ошибка соединения»

Question

Я установил на AWS кластер RabbitMQ с двумя узлами и включил HA, как описано here. Затем я настроил отображение балансировки эластичной нагрузки 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

Pinging admin port ничего не сделает, чтобы поддерживать соединение amqp.

Вам необходимо настроить биения сердца на достаточно низкое значение, чтобы предотвратить сброс балансировки нагрузки «холостого» соединения.

Похоже, вы используете Spring Boot (на основе имени свойства), и в настоящее время он не представляет requestedHeartbeats как собственность; вам нужно будет переопределить загрузочный завод, подключив свой собственный фасоль rabbitConnectionFactory с базовыми настройками заводских настроек соединения на основе кролика. Подробности см. На странице documentation.

При этом не существует преимуществ от использования балансировщика нагрузки с пружинным AMQP, поскольку он использует одно долгоживущее соединение. Подумайте о настройке двух экземпляров в свойстве addresses вместо этого, и он перейдет ко второму экземпляру, если первый из них недоступен.