文档章节

使用Spring Jms连接ActiveMQ 重连问题的一点心得

杨延庆
 杨延庆
发布于 2016/09/30 17:44
字数 1317
阅读 2759
收藏 37

我们现在做的Web项目遇到一个问题,项目启动时使用Failover方式连接某个ActiveMQ,brokerUrl类似

failOver:(tcp://xxx.xxx.xxx.xxx:61616),如果ActiveMQ启动正常,项目启动正常;

但如果ActiveMQ没有启动,项目将无法启动完成,一直卡在连接ActiveMQ阶段。

经过调查,发现是ActiveMQ提供的FailOver Transport方式导致的。

FailOver Transport有maxReconnectAttempts和startupMaxReconnectAttempts两个连接参数。

    前者是FailOver Transport最大的重连次数,后者是启动阶段的最大重连次数,由于我们原先的连接这两个参数没有设置值,它们都沿用默认值-1,表示无限次重连,直到连接成功为止。这样,只要我们在启动项目时不启动ActiveMQ,FailOver Transport将一直尝试重连,导致后续程序加载无法进行,从而项目启动卡住。

    找到了原因之后,我们在brokerUrl中设置startupMaxReconnectAttempts=2,形如:

    failOver:(tcp://xxx.xxx.xxx.xxx:61616)?startupMaxReconnectAttempts=2,这样启动时FailOverTransport仅仅重试2次,如果不成功,就不再尝试,继续加载后续程序,这样就解决了项目启动卡住的问题。

    然而还有一个问题没有解决,如果启动时没有连接ActiveMQ成功,在项目启动完成后,再启动ActiveMQ时,即使ActiveMQConnectionFactory一直重连ActiveMQ,直到连接成功。我们配置的DefaultMessageListenerContainer仍然会抛出以下异常:

ERROR DefaultMessageListenerContainer - Could not refresh JMS Connection for destination 'queue://xxxxxxx' - retrying in 5000 ms. Cause: The JMS connection has failed: Connection refused: connect.

   在StackOverflow上有人给出了解答,是因为DefaultMessageListenerContainer引用的

SingleConnectionFactory对象的reconnectOnException属性。

(http://stackoverflow.com/questions/17729066/spring-mq-jms-reconnect-configuration)

   这个属性的说明如下(基于Spring Framework 4.3.2 RELEASE版本)

   Specify whether the single Connection should be reset (to be subsequently renewed) when a JMSException is reported by the underlying Connection.

  这个属性指定是否在底层的连接(这里是指ActiveMQ连接)抛出JMSException的时候,对连接进行重置。

  它的默认值是false,意味着即使JMS异常抛出,SingleConnectionFactory自带的connection没有重置。

在调用getConnection()方法时,由于连接没有重置,this.connection != null,仍然返回旧的connection.

从而DefaultMessageListenerContainer引用的连接仍然失败连接,不断抛出异常

protected Connection getConnection() throws JMSException {
		synchronized (this.connectionMonitor) {
			if (this.connection == null) {
				initConnection();
			}
			return this.connection;
		}
	}

   当我们设置这个属性为true后,在抛出Jms异常时,SingleConnectionFactory对象绑定的ExceptionListner代理对象会回调SingleConnectionFactory对象的onException方法,将connection属性设置为null,这样在getConnection()方法时,会调用initConnection方法创建新的连接。这样就能保证ActiveMQ启动成功后,会有对应的成功连接被创建,从而使DefaultMessageListenerContainer引用到正常连接,不再抛出异常。

public class SingleConnectionFactory ......
{

    @Override
	public void onException(JMSException ex) {
		logger.warn("Encountered a JMSException - resetting the underlying JMS Connection", ex);
		resetConnection();
	}

    public void resetConnection() {
		synchronized (this.connectionMonitor) {
			if (this.connection != null) {
				closeConnection(this.connection);
			}
			this.connection = null;
		}
	}

    protected Connection getConnection() throws JMSException {
		synchronized (this.connectionMonitor) {
			if (this.connection == null) {
				initConnection();
			}
			return this.connection;
		}
	}

    public void initConnection() throws JMSException {
		if (getTargetConnectionFactory() == null) {
			throw new IllegalStateException(
					"'targetConnectionFactory' is required for lazily initializing a Connection");
		}
		synchronized (this.connectionMonitor) {
			if (this.connection != null) {
				closeConnection(this.connection);
			}
			this.connection = doCreateConnection();
			prepareConnection(this.connection);
			if (this.startedCount > 0) {
				this.connection.start();
			}
			if (logger.isInfoEnabled()) {
				logger.info("Established shared JMS Connection: " + this.connection);
			}
		}
	}
}

private class AggregatedExceptionListener implements ExceptionListener {

		final Set<ExceptionListener> delegates = new LinkedHashSet<ExceptionListener>(2);

		@Override
		public void onException(JMSException ex) {
			synchronized (connectionMonitor) {
				// Iterate over temporary copy in order to avoid ConcurrentModificationException,
				// since listener invocations may in turn trigger registration of listeners...
				for (ExceptionListener listener : new LinkedHashSet<ExceptionListener>(this.delegates)) {
					listener.onException(ex);
				}
			}
		}
	}



   上述代码基于Spring Framework 4.3.3.RELEASE版本

 具体的配置信息如下:

<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
        ...................
        <property name="reconnectOnException" value="true"/> 
</bean>

     上面的修改方案虽然可以解决ActiveMQ服务器未启动时应用启动卡住的问题,但是每次重置JMS连接时都会输出异常信息,这是因为重置连接时会将抛出的异常信息输出到日志。     

package org.springframework.jms.connection;

public class SingleConnectionFactory implements ConnectionFactory, QueueConnectionFactory,
		TopicConnectionFactory, ExceptionListener, InitializingBean, DisposableBean {
.............
public void onException(JMSException ex) {
		logger.warn("Encountered a JMSException - resetting the underlying JMS Connection", ex);
		resetConnection();
	}
.............
}

       如果不想查看重置连接时抛出的异常信息,可以在日志配置文件中设置日志输出级别为ERROR(以logback为例) 

<logger name="org.springframework.jms.connection" level="ERROR" />

       

       此外在ActiveMQ没有启动的情况下,如果在项目中为ActiveMQ消息队列配置了DefaultMessageListenerContainer对象,它将会持续刷新JMS连接,输出异常信息,默认的刷新间隔是由DefaultMessageListenerContainer对象的backOff属性对象的interval属性确定的

public class DefaultMessageListenerContainer extends AbstractPollingMessageListenerContainer {
...........
/**
	 * The default recovery interval: 5000 ms = 5 seconds.
*/
public static final long DEFAULT_RECOVERY_INTERVAL = 5000;
private BackOff backOff = new FixedBackOff(DEFAULT_RECOVERY_INTERVAL, Long.MAX_VALUE);

public void setRecoveryInterval(long recoveryInterval) {
		this.backOff = new FixedBackOff(recoveryInterval, Long.MAX_VALUE);
}
...........
}

public class FixedBackOff implements BackOff {

private long interval = 5000L;
........
public FixedBackOff(long interval, long maxAttempts) {
        this.interval = interval;
        this.maxAttempts = maxAttempts;
    }
........
}

    如果在定义DefaultMessageListenerContainer对象时没有设定recoveryInterval,那它将使用默认的5000ms初始化backoff对象。实际运行时DefaultMessageListenerContainer对象将使用5000ms(5s)这个时间间隔刷新JMS连接,输出连接异常信息。如果定义的DefaultMessageListenerContainer对象数目很多,日志中的连接异常信息量将会很大,我们需要调整这个时间间隔。
    可以在定义DefaultMessageListenerContainer对象时设置recoveryInterval(下面的例子将recoveryInterval设置为30000ms,大家可以根据自己的需求进行调整。) 

<bean id="xxxContainer" 
        class="org.springframework.jms.listener.DefaultMessageListenerContainer"> 
        ..................
		<property name="recoveryInterval" value="30000" />
</bean>

 

 

© 著作权归作者所有

共有 人打赏支持
杨延庆
粉丝 31
博文 47
码字总数 44076
作品 0
浦东
架构师
加载中

评论(1)

SandKing
SandKing
如果不使用spring怎么处理这个问题呢
Ubuntu下安装ActiveMQ

主要参考http://www.jmkg.co.uk/2010/08/31/installing-activemq-on-ubuntu/,略有补充 1. 下载安装包,建立activemq用户,安装 下载最新的activemq,写此文时为5.7。 cd /opt/tar xvzf xxx...

RuralHunter
2013/01/22
0
0
ActiveMQ集群方案(上)

版权声明:欢迎转载,但是看在我辛勤劳动的份上,请注明来源:http://blog.csdn.net/yinwenjie(未经允许严禁用于商业用途!) https://blog.csdn.net/yinwenjie/article/details/51124749 目...

yunlielai
04/15
0
0
ActiveMQ 嵌入Tomcat

在一些项目中,单独开启一个ActiveMQ,对于项目实施来说有时略显繁琐。所以我们将ActiveMQ内嵌到Tomcat,Tomcat启动同时就顺带启动了ActiveMQ。由此我们需要掌握三个个重要的知识点 ActiveMQ...

_herbert
08/03
0
0
ActiveMQ:No operations allowed after statement ...

ActiveMQ版本:5.5.1 记录人:@郑昀现象: 系统现象:部分消息发送失败,失败频率不正常。 日志信息:activemq.log 中一直有这样的错误日志: 看上去又是 mq broker 失去了数据库连接,但代码...

旁观者-郑昀
2012/11/10
0
0
ActiveMQ:Communications link failure问题以及解决办法

ActiveMQ版本:5.5.1 MQ 所使用的 MySQL 是 InnoDB存储引擎 记录人:@郑昀现象: 业务表面现象:无。系统现象:无。 日志信息:业务系统发送 MQ 消息时,下面这种错误日志断断续续地一直都有...

旁观者-郑昀
2012/11/10
0
0

没有更多内容

加载失败,请刷新页面

加载更多

OSChina 周六乱弹 —— 放假前期焦虑症晚期

Osc乱弹歌单(2018)请戳(这里) 【今日歌曲】 @andonny :分享Matteo的单曲《Panama》: 《Panama》- Matteo 手机党少年们想听歌,请使劲儿戳(这里) @新垣吉衣OSC :我发现只要去有小朋友...

小小编辑
15分钟前
21
4
wait()被notify()后,接着执行wait()后面的语句

wait()被notify()后,接着执行wait()后面的语句

noteman
46分钟前
1
0
Ubuntu集群-使用MAAS开始裸机安装

Ubuntu使用MAAS装机的七个步骤。 1、Setup your hardware You need one small server for MAAS and at least one server which can be managed with a BMC. It is recommended to have the M......

openthings
59分钟前
3
0
OSX | SafariBookmarksSyncAgent意外退出解决方法

1. 启动系统, 按住⌘-R不松手2. 在实用工具(Utilities)下打开终端,输入csrutil disable, 然后回车; 你就看到提示系统完整性保护(SIP: System Integrity Protection)已禁用3. 输入reboot回车...

云迹
今天
4
0
面向对象类之间的关系

面向对象类之间的关系:is-a、has-a、use-a is-a关系也叫继承或泛化,比如大雁和鸟类之间的关系就是继承。 has-a关系称为关联关系,例如企鹅在气候寒冷的地方生活,“企鹅”和“气候”就是关...

gackey
今天
4
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部