java - 设置最大消费者数量 jms solac

Question

我正在尝试使用 solace 作为代理，使用 jms 设置主题端点的最大消费者数量，因此为了增加负载，可以在 cloudfoundry 中启动应用程序的多个实例，并且多个订阅者可以使用同一主题的消息.

我尝试了以下设置的多种组合(setConcurrency()、setConcurrentConsumers()、setMaxConcurrentConsumers()，(20 作为任意高的数字)。从文档来看，我绝对需要使用 setMaxConcurrentConsumers() 并将其设置为适当的高值。

当我部署应用程序时，会创建主题端点，但是当我查看 solace 管理界面时，最大消费者计数始终为 1(如下所示:Queues -> Topic Endpoints -> select端点 -> 配置限制)，即使它应该是 20。因此第二个消费者无法连接。我不想每次部署应用程序时都手动设置它。

import javax.jms.*;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.connection.CachingConnectionFactory;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.listener.DefaultMessageListenerContainer;

@Configuration
public class ProducerConfiguration {

    private static final Log logger = LogFactory.getLog(SolaceController.class);

    @Value("${durable_subscription}")
    private String subscriptionName;

    @Value("${topic_name}")
    private String topic_name;

    @Autowired
    private ConnectionFactory connectionFactory;

    @Bean
    public JmsTemplate jmsTemplate() {
        CachingConnectionFactory ccf = new CachingConnectionFactory(connectionFactory);
        JmsTemplate jmst = new JmsTemplate(ccf);
        jmst.setPubSubDomain(true);
        return jmst;
    }

    @Bean
    public Session configureSession(ConnectionFactory connectionFactory) throws JMSException {
        return connectionFactory.createConnection().createSession(false, Session.AUTO_ACKNOWLEDGE);
    }


    private TextMessage lastReceivedMessage;

    public class SimpleMessageListener implements MessageListener {
        @Override
        public void onMessage(Message message) {

            if (message instanceof TextMessage) {
                lastReceivedMessage = (TextMessage) message;
                try {
                    logger.info("Received message : " + lastReceivedMessage.getText());
                } catch (JMSException e) {
                    logger.error("Error getting text of the received TextMessage: " + e);
                }
            } else {
                logger.error("Received message that was not a TextMessage: " + message);
            }
        }
    }

    @Bean
    public DefaultMessageListenerContainer orderMessageListenerContainer() {

        DefaultMessageListenerContainer lc = new DefaultMessageListenerContainer();
        lc.setConnectionFactory(connectionFactory);
        lc.setDestinationName(topic_name);
        lc.setMessageListener(new SimpleMessageListener());
        lc.setDurableSubscriptionName(subscriptionName);
        lc.setPubSubDomain(true);

        //tried multiple combinations here, also setting only setMaxConcurrentConsumers
        lc.setConcurrency("2-20");
        lc.setConcurrentConsumers(20);
        lc.setMaxConcurrentConsumers(20);

        lc.setSubscriptionDurable(true);
        lc.initialize();
        lc.start();
        return lc;
    }
}

Answer 1

我认为对于您的用例，您的消费者被队列困住了。请参阅https://solace.com/blog/topic-subscription-queues/

“...虽然多个消费者可以绑定(bind)到队列持久端点仅限于单个主题订阅。队列允许多个主题订阅以及主题通配符。”

如果您不想更改发布者，可以尝试“队列上的主题订阅”。也就是说，可以将队列配置为监听某个主题。然后您的消费者将从该队列中获取消息。