Spring AMQP устаревшие потребители для RabbitMQ на PCF - PullRequest
Новая
       45

Spring AMQP устаревшие потребители для RabbitMQ на PCF

0 голосов
/ 06 марта 2019

У нашей команды были некоторые проблемы с Spring AMQP при использовании соединителей облачного литейного производства. Всякий раз, когда в сети возникает некоторая нестабильность, происходит сбой соединения, и AMQP, похоже, пытается выполнить автоматическое восстановление, не удается выполнить автоматическое восстановление, а затем создает нового потребителя.

Проблема в том, что RabbitMQ, похоже, не отменяет регистрацию старых потребителей ... поэтому мы получаем 40 потребителей вместо обычного одного потребителя. И я думаю, что из-за количества недопустимых потребителей старые потребители получают некоторые сообщения, но не запускают наш код. Таким образом, у нас есть куча неупакованных сообщений, пока приложение не будет перезапущено, и не очистим список потребителей.

Я не уверен, что это неправильная конфигурация с нашей стороны или ошибка. Есть идеи?

Дополнительная информация ниже.

Часть наших журналов: 

a0b279ba-fb73-41cc-bdf9-b430cec52856 1 Mon Mar 04 2019 02:31:47.842 com.rabbitmq.client.TopologyRecoveryException: Caught an exception while recovering consumer amq.ctag-Dqik8vz9dSPiUzMbM4uGbw: connection is already closed due to connection error; cause: java.net.SocketException: Connection reset
a0b279ba-fb73-41cc-bdf9-b430cec52856 1 Mon Mar 04 2019 02:31:47.842     at java.lang.Thread.run(Thread.java:748) [na:1.8.0_162]
a0b279ba-fb73-41cc-bdf9-b430cec52856 1 Mon Mar 04 2019 02:31:47.842 2019-03-04 00:31:47.823 ERROR [Atlas Backend,,,] 23 --- [0.32.27.77:5672] c.r.c.impl.ForgivingExceptionHandler     : Caught an exception when recovering topology Caught an exception while recovering consumer amq.ctag-t7_vY47weZNYtWagl__pYA: connection is already closed due to connection error; cause: java.net.SocketException: Connection reset
a0b279ba-fb73-41cc-bdf9-b430cec52856 1 Mon Mar 04 2019 02:31:47.842     at com.rabbitmq.client.impl.AMQChannel.ensureIsOpen(AMQChannel.java:198) ~[amqp-client-4.0.3.jar!/:4.0.3]
a0b279ba-fb73-41cc-bdf9-b430cec52856 1 Mon Mar 04 2019 02:31:47.842 2019-03-04 00:31:47.823 ERROR [Atlas Backend,,,] 23 --- [0.32.27.77:5672] c.r.c.impl.ForgivingExceptionHandler     : Caught an exception when recovering topology Caught an exception while recovering consumer amq.ctag-AK1YOpN0_E2KVwSi4fzOaw: connection is already closed due to connection error; cause: java.net.SocketException: Connection reset
a0b279ba-fb73-41cc-bdf9-b430cec52856 1 Mon Mar 04 2019 02:31:47.842     at com.rabbitmq.client.impl.AMQConnection.notifyRecoveryCanBeginListeners(AMQConnection.java:693) [amqp-client-4.0.3.jar!/:4.0.3]
a0b279ba-fb73-41cc-bdf9-b430cec52856 1 Mon Mar 04 2019 02:31:47.842     at com.rabbitmq.client.impl.AMQConnection.doFinalShutdown(AMQConnection.java:687) [amqp-client-4.0.3.jar!/:4.0.3]
a0b279ba-fb73-41cc-bdf9-b430cec52856 1 Mon Mar 04 2019 02:31:47.842     at com.rabbitmq.client.impl.AMQConnection$MainLoop.run(AMQConnection.java:577) [amqp-client-4.0.3.jar!/:4.0.3]
a0b279ba-fb73-41cc-bdf9-b430cec52856 1 Mon Mar 04 2019 02:31:47.842     at java.lang.Thread.run(Thread.java:748) [na:1.8.0_162]
a0b279ba-fb73-41cc-bdf9-b430cec52856 1 Mon Mar 04 2019 02:31:47.842 2019-03-04 00:31:47.822 ERROR [Atlas Backend,,,] 23 --- [0.32.27.77:5672] c.r.c.impl.ForgivingExceptionHandler     : Caught an exception when recovering topology Caught an exception while recovering consumer amq.ctag-vrWhRdhk3ejrXo2-ImjXzA: connection is already closed due to connection error; cause: java.net.SocketException: Connection reset
a0b279ba-fb73-41cc-bdf9-b430cec52856 1 Mon Mar 04 2019 02:31:47.842     at com.rabbitmq.client.impl.AMQChannel.rpc(AMQChannel.java:244) ~[amqp-client-4.0.3.jar!/:4.0.3]
a0b279ba-fb73-41cc-bdf9-b430cec52856 1 Mon Mar 04 2019 02:31:47.842     at java.lang.Thread.run(Thread.java:748) [na:1.8.0_162]
a0b279ba-fb73-41cc-bdf9-b430cec52856 1 Mon Mar 04 2019 02:31:47.842     at java.lang.Thread.run(Thread.java:748) [na:1.8.0_162]
a0b279ba-fb73-41cc-bdf9-b430cec52856 1 Mon Mar 04 2019 02:31:47.842     at com.rabbitmq.client.impl.recovery.RecordedConsumer.recover(RecordedConsumer.java:60) ~[amqp-client-4.0.3.jar!/:4.0.3]
a0b279ba-fb73-41cc-bdf9-b430cec52856 1 Mon Mar 04 2019 02:31:47.842     ... 7 common frames omitted
a0b279ba-fb73-41cc-bdf9-b430cec52856 1 Mon Mar 04 2019 02:31:47.842     at com.rabbitmq.client.impl.recovery.AutorecoveringConnection.recoverConsumers(AutorecoveringConnection.java:657) [amqp-client-4.0.3.jar!/:4.0.3]
a0b279ba-fb73-41cc-bdf9-b430cec52856 1 Mon Mar 04 2019 02:31:47.842     at com.rabbitmq.client.impl.recovery.AutorecoveringConnection.access$000(AutorecoveringConnection.java:53) [amqp-client-4.0.3.jar!/:4.0.3]
a0b279ba-fb73-41cc-bdf9-b430cec52856 1 Mon Mar 04 2019 02:31:47.842     at com.rabbitmq.client.impl.recovery.AutorecoveringConnection.recoverConsumers(AutorecoveringConnection.java:657) [amqp-client-4.0.3.jar!/:4.0.3]
a0b279ba-fb73-41cc-bdf9-b430cec52856 1 Mon Mar 04 2019 02:31:47.842     at com.rabbitmq.client.impl.AMQConnection$MainLoop.run(AMQConnection.java:577) [amqp-client-4.0.3.jar!/:4.0.3]
a0b279ba-fb73-41cc-bdf9-b430cec52856 1 Mon Mar 04 2019 02:31:47.842     at com.rabbitmq.client.impl.ChannelN.basicConsume(ChannelN.java:1242) ~[amqp-client-4.0.3.jar!/:4.0.3]
a0b279ba-fb73-41cc-bdf9-b430cec52856 1 Mon Mar 04 2019 02:31:47.842     at com.rabbitmq.client.impl.ChannelN.basicConsume(ChannelN.java:1242) ~[amqp-client-4.0.3.jar!/:4.0.3]
a0b279ba-fb73-41cc-bdf9-b430cec52856 1 Mon Mar 04 2019 02:31:47.842     at com.rabbitmq.client.impl.recovery.AutorecoveringConnection.recoverConsumers(AutorecoveringConnection.java:657) [amqp-client-4.0.3.jar!/:4.0.3]
a0b279ba-fb73-41cc-bdf9-b430cec52856 1 Mon Mar 04 2019 02:31:47.842     at com.rabbitmq.client.impl.AMQConnection.doFinalShutdown(AMQConnection.java:687) [amqp-client-4.0.3.jar!/:4.0.3]
a0b279ba-fb73-41cc-bdf9-b430cec52856 1 Mon Mar 04 2019 02:31:47.842     at com.rabbitmq.client.impl.AMQConnection.notifyRecoveryCanBeginListeners(AMQConnection.java:693) [amqp-client-4.0.3.jar!/:4.0.3]
a0b279ba-fb73-41cc-bdf9-b430cec52856 1 Mon Mar 04 2019 02:31:47.842     at com.rabbitmq.client.impl.AMQChannel.ensureIsOpen(AMQChannel.java:198) ~[amqp-client-4.0.3.jar!/:4.0.3]
a0b279ba-fb73-41cc-bdf9-b430cec52856 1 Mon Mar 04 2019 02:31:47.842     at com.rabbitmq.client.impl.recovery.RecordedConsumer.recover(RecordedConsumer.java:60) ~[amqp-client-4.0.3.jar!/:4.0.3]
a0b279ba-fb73-41cc-bdf9-b430cec52856 1 Mon Mar 04 2019 02:31:47.842     at com.rabbitmq.client.impl.recovery.RecordedConsumer.recover(RecordedConsumer.java:60) ~[amqp-client-4.0.3.jar!/:4.0.3]
a0b279ba-fb73-41cc-bdf9-b430cec52856 1 Mon Mar 04 2019 02:31:47.842     at com.rabbitmq.client.impl.recovery.AutorecoveringConnection.recoverConsumers(AutorecoveringConnection.java:673) [amqp-client-4.0.3.jar!/:4.0.3]
a0b279ba-fb73-41cc-bdf9-b430cec52856 1 Mon Mar 04 2019 02:31:47.842     at com.rabbitmq.client.impl.ChannelN.basicConsume(ChannelN.java:1242) ~[amqp-client-4.0.3.jar!/:4.0.3]
a0b279ba-fb73-41cc-bdf9-b430cec52856 1 Mon Mar 04 2019 02:31:47.842     at java.lang.Thread.run(Thread.java:748) [na:1.8.0_162]

Мы регистрируем наших потребителей, используя аннотацию @RabbitListener 

    @RabbitListener(id = CONSUMER_ID, queues = "${CommonInternalEndPoint}", containerFactory = "rabbitLocalContainerFactory")
    protected void process(Object messageObject) {

Поскольку мы подключаемся к нескольким экземплярам rabbit mq, нам нужно объявить фабрику соединений вручную. Наша фабрика соединений основана на соединителях облачного литейного производства: 

@Configuration
@Profile("cloud")
public class RabbitCloudConfiguration {

    @Bean
    public Cloud cloud(){
        return new CloudFactory().getCloud();
    }

    @Bean
    @Primary
    public ConnectionFactory connectionFactory(){
        return cloud().getSingletonServiceConnector(ConnectionFactory.class, null);
    }

    @Bean("rabbitLocalContainerFactory")
    public SimpleRabbitListenerContainerFactory rabbitLocalContainerFactory(){
        SimpleRabbitListenerContainerFactory containerFactory = new SimpleRabbitListenerContainerFactory();
        containerFactory.setConnectionFactory(connectionFactory());
        containerFactory.setAutoStartup(true);
        return containerFactory;
    }
}

1 Ответ

1 голос
/ 07 марта 2019

Мы используем Spring Boot 1.5.9, Framework 4.3.6, AMQP 1.7.4, RabbitMQ AMQP Client 4.0.3

Начиная с версии 2.0, Spring AMQP отключает automaticRecoveryEnabled на целевом RabbitMQ ConnectionFactory: https://docs.spring.io/spring-amqp/docs/2.1.4.RELEASE/reference/#auto-recovery

Поскольку используемая версия основана на RabbitMQ ConnectionFactory, предоставляемом Cloud Connectors, вам необходимо явно отключить его: 

@Bean
@Primary
public ConnectionFactory connectionFactory(){
    ConnectionFactory connectionFactory = cloud().getSingletonServiceConnector(ConnectionFactory.class, null);
    ((CachingConnectionFactory) connectionFactory).getRabbitConnectionFactory().setAutomaticRecoveryEnabled(false);
    return connectionFactory;
}

И ваше решение будет основано на механизме восстановления, предоставляемом Spring AMQP.

...