问题描述
我对 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}
解决方法
暂无找到可以解决该程序问题的有效方法,小编努力寻找整理中!
如果你已经找到好的解决方法,欢迎将解决方案带上本链接一起发送给小编。
小编邮箱:dio#foxmail.com (将#修改为@)