在ActiveMQ中使用SingleConnectionFactory遇到的坑

我们在生产环境使用了ActiveMQ作为消息中间件,消息中间件连接到数据库对消息进行持久化。最近发生了一个奇怪的事情,消费者端的生产日志总是报如下错误:TheJMSconnectionhas
强烈推介IDEA2021.1.3破解激活,IntelliJ IDEA 注册码,2021.1.3IDEA 激活码  

我们在生产环境使用了ActiveMQ作为消息中间件,消息中间件连接到数据库对消息进行持久化。

最近发生了一个奇怪的事情,消费者端的生产日志总是报如下错误:

The JMS connection has failed: java.io.EOFException

Successfully refreshed JMS Connection

这个日志在生产环境大量重复,第一个错误是EOFException,当一个连接的远端主动关闭连接时,本端会接收到这个异常。

第二行的消息咋一看,是“Successfully refreshed JMS Connection”,理论上此时连接已经恢复,但是消费者依然无法获取信息。

将这个问题放到Baidu,Bing和Google上搜索,也看到了其他用户遇到过类似情况,但是仔细研究一下发现并不是我们遇到的情况,他们提供的解决方案也无法解决我们的问题。

 

就在问题要陷入僵局的时候,我们发现:

1. 在错误消息刷出来之前,ActiveMQ报了错,疑似它使用的持久化数据库中途挂掉,且被重启了

2. 同一个MQ有几个组件作为消费端,但是只有这个组件刷错误日志,其他组件正常恢复了连接

 

初步分析是因为数据库挂掉导致连接失效,因为发现MQ日志中,数据库连接报错之后很短的时间间隔内客户端就开始刷日志。

很奇怪的是其他几个组件没有问题,所以我们对比了这几个组件之间的配置文件差异,发现只有这个组件使用了SingleConnectionFactory,其他组件使用的都是CachingConnectionFactory,所以怀疑问题出在了这个配置上。

对比了下这两个类的源代码,这是CachingConnectionFactory.java的注释部分和第一行代码:

/**
 * {@link SingleConnectionFactory} subclass that adds {@link javax.jms.Session}
 * caching as well {@link javax.jms.MessageProducer} caching. This ConnectionFactory
 * also switches the {@link #setReconnectOnException "reconnectOnException" property}
 * to "true" by default, allowing for automatic recovery of the underlying Connection.
 *
......
 *
 * @author Juergen Hoeller
 * @since 2.5.3
 */

public class CachingConnectionFactory extends SingleConnectionFactory {
......

只听到从山间传来架构君的声音:
绣阁银屏一夜秋。有谁来对上联或下联?

大家可以注意到,这个类就是扩展了SingleConnectionFactory类,并且将reconnectOnException设置为true

至此,问题已经有一点眉目了,如果连接出现异常,通过reconnectOnException决定是否reconnect(重连接),这个属性在SingleConnectionFactory默认设置为false的(可以参见它的代码,默认设置为false),但是在CachingConnectionFactory中设置为true,这就是为何连接失效了,客户端却没能重新连上的原因。

进一步检查日志中消息“Successfully refreshed JMS Connection”的来源,可以进一步印证我们的看法:

 请查看DefaultMessageListenerContainer.java:

此代码由Java架构师必看网-架构君整理
package org.springframework.jms.listener; ... public class DefaultMessageListenerContainer extends AbstractPollingMessageListenerContainer { ... protected void refreshConnectionUntilSuccessful() { while (true) { if (this.isRunning()) { try { if (this.sharedConnectionEnabled()) { this.refreshSharedConnection(); } else { Connection con = this.createConnection(); JmsUtils.closeConnection(con); } this.logger.info("Successfully refreshed JMS Connection"); } catch (Exception var3) { ... } ... }

可以看到,上面代码中的消息,正是我们在日志中反复看到的信息,而通过检查refreshSharedConnection()和createConnection(),我们发现:

refreshSharedConnection()调用链:

AbstractJmsListeningContainer.refreshSharedConnection()
AbstractJmsListeningContainer.createSharedConnection()
JmsAccessor.createConnection()
JmsAccessor.getConnectionFactory().createConnection()

createConnection()调用链:

此代码由Java架构师必看网-架构君整理
JmsAccessor.createConnection()
JmsAccessor.getConnectionFactory().createConnection()

可以看出,这两个分支最后都是到连接工厂中调用createConnection(),查看下代码:

package org.springframework.jms.connection;

public class SingleConnectionFactory
...
    public Connection createConnection() throws JMSException {
        Object var1 = this.connectionMonitor;
        synchronized (this.connectionMonitor) {
            if (this.connection == null) {
                this.initConnection();
            }

            return this.connection;
        }
    }
...
}

使用SingleConnectionFactory时,如果连接对象connection不为空,即使此时连接失效,依然不会进入initConnection,所以虽然connection返回了非null值,但是这个连接其实是坏的

至此,原因查明,将SingleConnectionFactory改为CachingConnectionFactory后,此问题消失,当客户端发现连接失效后,能够主动连接到MQ服务器

 

思考:为何设置了“reconnectOnException”属性后,就能够自动重连了呢?

找到代码中使用了这个变量的位置:

package org.springframework.jms.connection;

public class SingleConnectionFactory
...
    protected void prepareConnection(Connection con) throws JMSException {
        if (this.getClientId() != null) {
            con.setClientID(this.getClientId());
        }

        if (this.getExceptionListener() != null || this.isReconnectOnException()) {
            ExceptionListener listenerToUse = this.getExceptionListener();
            if (this.isReconnectOnException()) {
                listenerToUse = new InternalChainedExceptionListener(this, (ExceptionListener) listenerToUse);
            }

            con.setExceptionListener((ExceptionListener) listenerToUse);
        }

    }
...
}

代码中添加了一个异常监听器,此监听器触发的代码为:

package org.springframework.jms.connection;

public class SingleConnectionFactory
{
...

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

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

            this.target = null;
            this.connection = null;
        }
    }
...
}

可见,当发生异常时,异常监听器调用了resetConnection()函数,此函数会将connection设置为null,然后DefaultMessageListenerContainer中的监视线程经过一段时间即能重新建立此连接

架构君码字不易,如需转载,请注明出处:https://javajgs.com/archives/42484
0

发表评论