Блог

Я столкнулся с одной проблемой, когда сообщение в Acpache ActiveMQ удалено из очереди в консоли, но не вызывается метод-потребитель, у которого есть @JmsListener, хотя количество потребителей показывает 7.

Код конфигурации:

 @Configuration
@EnableJms
public class JMSConfiguration {
    private static final String DEFAULT_BROKER_URL = "tcp://localhost:61616";
    private static final long DEFAULT_RECEIVE_TIMEOUT = 1000L;
    private static final String DEFAULT_CONCURRENCY = "25";

    @Bean
    public ConnectionFactory connectionFactory() {
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
        connectionFactory.setBrokerURL(DEFAULT_BROKER_URL);
        return connectionFactory;
    }

    @Bean
    public DefaultJmsListenerContainerFactory jmsListenerContainerFactory() {
        DefaultJmsListenerContainerFactory listenerFactory = new DefaultJmsListenerContainerFactory();
        listenerFactory.setConnectionFactory(connectionFactory());
        listenerFactory.setReceiveTimeout(DEFAULT_RECEIVE_TIMEOUT);
        listenerFactory.setConcurrency(DEFAULT_CONCURRENCY);
        return listenerFactory;
    }

    @Bean
    public JmsTemplate jmsTemplate() {
        JmsTemplate template = new JmsTemplate();
        template.setConnectionFactory(connectionFactory());
        return template;
    }
}
  

Код прослушивателя:

 @JmsListener(destination = Constants.EXPORT_QUEUE)
    public void onExportMessageReceive(Message message) {
        if (message instanceof Message) {
}
}
  

Я внес одно изменение в использование spring-jms версии 4.3.22-RELEASE.

Теперь это работает следующим образом: изначально в приложении определено 3 числа потребителей и максимум 25. И теперь, если будет создано больше сообщений, скажем, 25, тогда потребителей останется только 3, и количество потребителей будет постепенно увеличиваться на основе ожидающих сообщений. Он может увеличиться до 13, но не создаст максимальное количество потребителей до 25 в соответствии с определением. Из-за этого порядок может измениться. например: Сообщение 1 Consumer1 (выполняется) Сообщение 2 Consumer2 (выполняется) Сообщение 3 Consumer3 (выполняется)

Теперь, если эти сообщения находятся в процессе обработки и появились новые сообщения, они будут ждать потребителей, как указано ниже Message4 Consumer1 (ожидание) Message5 Consumer2 (ожидание) Message6 Consumer3 (ожидание)

Теперь, поскольку эти сообщения 4-6 находятся в состоянии ожидания и продолжают появляться новые сообщения, что вынуждает систему создавать новых потребителей, тогда эти сообщения будут использованы первыми. Сообщение 7 Consumer4 (выполняется) Сообщение 8 Consumer5 (выполняется)

Сначала обрабатываются сообщения 7-8, затем сообщения 4-6. Но они будут использованы, как только потребители получат свободу.