首页 文章

DefaultJmsListenerContainerFactory和并行连接未关闭

提问于
浏览
1

我使用Spring 4.x的DefaultJmsListenerContainerFactory来连接到ActiveMQ队列,使用@JmsListener处理来自该队列的消息,然后将消息推送到同一ActiveMQ代理上的主题 .

我正在为使用者/监听器和 生产环境 者使用单个缓存连接工厂,并且我将缓存使用者设置为false,这样我就可以缓存 生产环境 者,而不是消费者 . 我还将并发性设置为1-3,我希望在应用程序启动时队列中至少有1个消费者,随着消息的增加,消费者的数量将达到3.但是,作为数量消息减少了,我期待消费者的数量也会回落到1 . 但是,如果我查看线程(defaultmessagelistenercontainer-2/3),它们处于等待状态,并且它们不会关闭 . 当负载消退时,预计消费者数量也会被关闭,这不是预期的行为吗?请参阅下面的配置,如果这种行为不是开箱即用,请告诉我,如果我需要添加一些内容,以便按照上面的说明进行操作 .

ApplicationContext.java

@Bean
public DefaultJmsListenerContainerFactory jmsListenerContainerFactory() throws Throwable {
    DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();

    factory.setConnectionFactory(connectionFactory());
    factory.setConcurrency(environment.getProperty("jms.connections.concurrent"));
    factory.setSessionTransacted(environment.getProperty("jms.connections.transacted", Boolean.class));
    return factory;
}

@Bean
public CachingConnectionFactory connectionFactory(){
    RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
    redeliveryPolicy.setInitialRedeliveryDelay(environment.getProperty("jms.redelivery.initial-delay", Long.class));
    redeliveryPolicy.setRedeliveryDelay(environment.getProperty("jms.redelivery.delay", Long.class));
    redeliveryPolicy.setMaximumRedeliveries(environment.getProperty("jms.redelivery.maximum", Integer.class));
    redeliveryPolicy.setUseExponentialBackOff(environment.getProperty("jms.redelivery.use-exponential-back-off", Boolean.class));
    redeliveryPolicy.setBackOffMultiplier(environment.getProperty("jms.redelivery.back-off-multiplier", Double.class));

    ActiveMQConnectionFactory activeMQ = new ActiveMQConnectionFactory(environment.getProperty("jms.queue.username"), environment.getProperty("jms.queue.password"), environment.getProperty("jms.broker.endpoint"));
    activeMQ.setRedeliveryPolicy(redeliveryPolicy);
    activeMQ.setPrefetchPolicy(prefetchPolicy());

    CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(activeMQ);
    cachingConnectionFactory.setCacheConsumers(environment.getProperty("jms.connections.cache.consumers", Boolean.class));
    cachingConnectionFactory.setSessionCacheSize(environment.getProperty("jms.cache.size", Integer.class));
    return cachingConnectionFactory;
}

@Bean
public JmsMessagingTemplate jmsMessagingTemplate(){
    ActiveMQTopic activeMQ = new ActiveMQTopic(environment.getProperty("jms.queue.out"));

    JmsMessagingTemplate jmsMessagingTemplate = new JmsMessagingTemplate(connectionFactory());
    jmsMessagingTemplate.setDefaultDestination(activeMQ);

    return jmsMessagingTemplate;
}

application.properties

jms.connections.concurrent=1-3
jms.connections.prefetch=1000
jms.connections.transacted=true
jms.connections.cache.consumers=false
jms.redelivery.initial-delay=1000
jms.redelivery.delay=1000
jms.redelivery.maximum=5
jms.redelivery.use-exponential-back-off=true
jms.redelivery.back-off-multiplier=2
jms.cache.size=3
jms.queue.in=in.queue
jms.queue.out=out.queue
jms.broker.endpoint=failover:(tcp://localhost:61616)

1 回答

相关问题