Блог

Я использую JmsListener Spring boot для удаления сообщений из очереди IBM MQ в сеансе транзакций.

Проблема возникла ,когда его пропускная способность была ограничена примерно 50 транзакциями в секунду(удаление из очереди MQ, обработка и сохранение в базе данных).

ЧТОБЫ повысить пропускную способность, мы увеличили его параллелизм, и это действительно помогло. Но теперь мы видим, что во время перезапуска приложения одно и то же сообщение используется несколькими одновременными слушателями.(К этому я пришел, зарегистрировав имена потоков).

Это происходит не при больших нагрузках, а во время перезапуска приложения, когда сообщения ожидают удаления из очереди.

Существует ли какой-либо другой способ достижения высокой пропускной способности при сохранении максимальной скорости полного потребления с помощью транзакционного JsmListener?

     @Bean
public PlatformTransactionManager transactionManager() {
    JmsTransactionManager transactionManager = new JmsTransactionManager();
    transactionManager.setConnectionFactory(jmsConnectionFactory());
    return transactionManager;
}
    @Bean
public ConnectionFactory jmsConnectionFactory(){
    MQConnectionFactory cf = new MQConnectionFactory();
    try {
        cf.setStringProperty(WMQConstants.WMQ_HOST_NAME, host);
        cf.setIntProperty(WMQConstants.WMQ_PORT, Integer.parseInt(port));
        cf.setStringProperty(WMQConstants.WMQ_CHANNEL, channel);
        cf.setIntProperty(WMQConstants.WMQ_CONNECTION_MODE, WMQConstants.WMQ_CM_CLIENT);
        cf.setStringProperty(WMQConstants.WMQ_QUEUE_MANAGER, queueManager);
        cf.setStringProperty(WMQConstants.WMQ_APPLICATIONNAME, "JmsGet (JMS)");
        cf.setBooleanProperty(WMQConstants.USER_AUTHENTICATION_MQCSP, true);
        cf.setStringProperty(WMQConstants.USERID, user);
        cf.setStringProperty(WMQConstants.PASSWORD, password);
        cf.setStringProperty(WMQConstants.WMQ_SSL_CIPHER_SUITE, null);
    } catch (JMSException jmsException) {
        log.error(jmsException.toString());
    }
    return cf;
}
    @Bean
public JmsListenerContainerFactory<?> listenerContainerFactory(ConnectionFactory connectionFactory,
                                                                        DefaultJmsListenerContainerFactoryConfigurer configurer) {
    DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
    factory.setConcurrency("2");//Added to increase throughput
    factory.setTransactionManager(transactionManager());
    factory.setSessionTransacted(true);
    configurer.configure(factory, connectionFactory);
    return factory;
}
@JmsListener(containerFactory = "listenerContainerFactory", destination = "${input_qname}")
public void onMessage(String message) throws Exception{
//if exception -> throw e; 
}