java - ActiveMQ 出列未消费消息

Question

我对 ActiveMQ 和消息代理很陌生。我面临一个问题，我想从你那里得到一些提示和信息。

所以我有多个队列，每个队列只有一个侦听器，一切正常，直到连接丢失。连接建立后，队列和订阅者之间的连接也建立了，但重新连接后，未消费的消息不再发送。

如何重新发送尚未使用的消息？我正在使用 Spring 并且我正在寻找一种配置以使其工作，但没有找到任何配置，之后我一直在寻找解决方法或其他东西，但仍然无法找到解决方案。任何与我的问题相关的信息、解决方案、链接都会非常有帮助！

这是我的 JMSConfig 类：

@Configuration
class JMSConfig {

    private final String BROKER_URL = ConfigurationManager.getConfig().getString("broker.url");
    private final String BROKER_USERNAME = ConfigurationManager.getConfig().getString("broker.username");
    private final String BROKER_PASSWORD = ConfigurationManager.getConfig().getString("broker.password");

    private final long SENDER_RECEIVE_TIMEOUT = ConfigurationManager.getConfig().getLong("sender.receive.timeout");
    private final long SENDING_TIME_TO_LIVE = ConfigurationManager.getConfig().getLong("sending.time.to.live");

    private final String RECEIVER_LISTENER_CONCURRENCY = ConfigurationManager.getConfig().getString("receiver.listener.concurrency");

    private final boolean USE_POOLED_CONNECTION_FACTORY = ConfigurationManager.getConfig().getBoolean("use.pooled.connection.factory");
    private final int POOL_MAX_CONNECTIONS = ConfigurationManager.getConfig().getInt("pool.max.connections");

    private final boolean USE_TOPICS = ConfigurationManager.getConfig().getBoolean("use.topics");

    private final static boolean OVERRIDE_REDELIVERY = false;

    static {
        ConfigurationManager.registerConfiguration("messaging", ConfigurationPriority.LOWEST);
    }

    @Bean
    public ConnectionFactory connectionFactory() {
        LoggerUtilities.logger().info("Registering connectionFactory");

        // Setup the JMS connection
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
        connectionFactory.setBrokerURL(BROKER_URL);
        connectionFactory.setPassword(BROKER_USERNAME);
        connectionFactory.setUserName(BROKER_PASSWORD);
        // Does not work with spring.activemq.packages.trust-all=true so we need to setup manually
        // Avoiding "This class is not trusted to be serialized as ObjectMessage payload"
        connectionFactory.setTrustAllPackages(true);

        if (OVERRIDE_REDELIVERY) {
            overrideRedelivery(connectionFactory);
        }

        if (USE_POOLED_CONNECTION_FACTORY) {
            // Using PooledConnectionFactory will ensure that on the ActiveMQ server does not get "Exceeded the maximum number of allowed client connections."
            // This setup is also found in /apache-activemq-5.15.6/conf/activemq.xml > maximumConnections
            PooledConnectionFactory pooledConnectionFactory = new PooledConnectionFactory(connectionFactory);
            pooledConnectionFactory.setMaxConnections(POOL_MAX_CONNECTIONS);
            return pooledConnectionFactory;
        }

        return connectionFactory;
    }

    @Bean
    public JmsTemplate jmsTemplate() {
        LoggerUtilities.logger().info("Registering jmsTemplate");

        JmsTemplate template = new JmsTemplate();
        template.setConnectionFactory(connectionFactory());

        // Qos needs to be enabled when getDeliveryMode(), getPriority(), getTimeToLive() is needed
        template.setExplicitQosEnabled(true);
        template.setDeliveryPersistent(false);
        template.setSessionAcknowledgeMode(Session.AUTO_ACKNOWLEDGE);
        template.setTimeToLive(SENDING_TIME_TO_LIVE);
        template.setReceiveTimeout(SENDER_RECEIVE_TIMEOUT);

        if (USE_TOPICS) {
            template.setPubSubDomain(true);
        }

        return template;
    }

    @Bean
    public DefaultJmsListenerContainerFactory jmsListenerContainerFactory() {
        LoggerUtilities.logger().info("Registering jmsListenerContainerFactory");

        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();

        if (USE_TOPICS) {
            factory.setPubSubDomain(true);
        }

        factory.setConnectionFactory(connectionFactory());
        factory.setConcurrency(RECEIVER_LISTENER_CONCURRENCY);

        factory.setErrorHandler(throwable -> {
            String exceptionMessage = "[RECEIVER] Error on the listener = " + throwable.getMessage();
            LoggerUtilities.logger().warn(exceptionMessage);
        });

        return factory;
    }

    @Bean
    public MessageConverter jacksonJmsMessageConverter() {
        LoggerUtilities.logger().info("Registering jacksonJmsMessageConverter");

        // Serializing message content to json
        MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
        converter.setTargetType(MessageType.TEXT);
        converter.setTypeIdPropertyName("_type");
        return converter;
    }

    private void overrideRedelivery(ActiveMQConnectionFactory connectionFactory) {
        LoggerUtilities.logger().info("Registering overrideRedelivery");

        // Overriding redelivery policy
        RedeliveryPolicy policy = connectionFactory.getRedeliveryPolicy();
        policy.setInitialRedeliveryDelay(0);
        policy.setBackOffMultiplier(0);
        policy.setUseExponentialBackOff(false);
        policy.setMaximumRedeliveries(RedeliveryPolicy.NO_MAXIMUM_REDELIVERIES);
    }
}

以下是属性：

spring.jms.listener.acknowledge-mode= auto
spring.jms.listener.auto-startup= true
spring.jms.listener.concurrency= 3
spring.jms.listener.max-concurrency= 9
spring.jms.listener.pub-sub-domain= false
spring.jms.listener.receive-timeout=${BROKER_TIMEOUT}
spring.jms.template.delivery-mode= non_persistent
spring.jms.template.priority= 5
spring.jms.template.qos-enabled= true
spring.jms.template.receive-timeout= ${BROKER_TIMEOUT}
spring.jms.template.time-to-live= ${BROKER_TIMEOUT}

标签： javaactivemqspring-jms