ActiveMQ 5.10.2: [org.apache.activemq.broker.TransportConnection] TransportConnection (triggerStartAsyncNetworkBridgeCreation: 无法添加连接

Question

我们使用 Camel 2.9.8 和 ActiveMQ，最近将 ActiveMq 从 5.5.1 升级到 5.10.2（最后一个 JDK6 版本，因为我们在 JBoss EAP 6 上运行）。我们的设置是一个简单的“代理网络”（使用发现），多个节点通过 tcp:// 和 vm:// 连接器连接。

一切正常，但经过几天 100% 的工作后，我在日志中突然收到“无法添加连接”警告。更准确地说：

2019-12-05 09:31:01,623 WARN  [org.apache.activemq.broker.TransportConnection] TransportConnection (triggerStartAsyncNetworkBridgeCreation: remoteBroker=tcp://nlheevpat01pha.x.global/10.32.50.163:61612@41557, localBroker= vm://nlheevpat02pha_prd_01#6262) Failed to add Connection nlheevpat02pha_prd_01->nlheevpat01pha_prd_01-55572-1575391926556-6265:1

我们从未在旧 5.5.1 版本中收到警告。 activeMQ 队列仍然正常运行，所有预期的连接都显示在 activeMQ Web 管理控制台的“连接”选项卡下。

我们的配置如下所示

<bean id="placeholderConfig"
    class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer" />

<!-- shutdown hook is disabled as RAR classloader may be gone at shutdown -->
<amq:broker xmlns="http://activemq.apache.org/schema/core"
    brokerName="${activemq.broker.name}"
    useJmx="true"
    useShutdownHook="false"
    networkConnectorStartAsync="true"
    dataDirectory="${activemq.data.dir}">

    <amq:destinationPolicy>
        <amq:policyMap>
            <amq:policyEntries>
                <amq:policyEntry queue=">" producerFlowControl="true" memoryLimit="500mb">
                    <amq:deadLetterStrategy>
                        <amq:individualDeadLetterStrategy queuePrefix="DLQ." />
                    </amq:deadLetterStrategy>
                </amq:policyEntry>
                <amq:policyEntry topic=">" producerFlowControl="true" memoryLimit="300mb">
                    <amq:deadLetterStrategy>
                        <amq:individualDeadLetterStrategy queuePrefix="DLQ." />
                    </amq:deadLetterStrategy>
                </amq:policyEntry>


                <amq:policyEntry queue="Service.Request" producerFlowControl="true" memoryLimit="3 gb">
                    <amq:dispatchPolicy>
                        <amq:roundRobinDispatchPolicy />
                    </amq:dispatchPolicy>
                </amq:policyEntry>

            </amq:policyEntries>

        </amq:policyMap>
    </amq:destinationPolicy>

    <amq:managementContext>
        <!-- use appserver provided context instead of creating one, for jboss use: -Djboss.platform.mbeanserver -->
        <amq:managementContext createConnector="false" />
    </amq:managementContext>

    <amq:networkConnectors>
        <amq:networkConnector
            uri="multicast://${activemq.network.multicast.address}:${activemq.network.multicast.port}?group=${activemq.network.multicast.group}"
            userName="${activemq.auth.system.username}"
            password="${activemq.auth.system.password}"
            conduitSubscriptions="false"
            decreaseNetworkConsumerPriority="true"
        />
    </amq:networkConnectors>

    <amq:persistenceAdapter>
        <amq:kahaDB
            directory="${activemq.data.dir}/kahadb"
            checkForCorruptJournalFiles="true"
            checksumJournalFiles="true"
        />
    </amq:persistenceAdapter>

    <amq:plugins>
        <amq:simpleAuthenticationPlugin>
            <amq:users>
                <amq:authenticationUser
                    username="${activemq.auth.system.username}"
                    password="${activemq.auth.system.password}"
                    groups="users"
                />
            </amq:users>
        </amq:simpleAuthenticationPlugin>

        <amq:authorizationPlugin>
            <amq:map>
                <amq:authorizationMap>
                    <amq:authorizationEntries>
                        <amq:authorizationEntry  queue=">" read="users" write="users" admin="users" />
                        <amq:authorizationEntry  topic=">" read="users" write="users" admin="users" />
                    </amq:authorizationEntries>
                </amq:authorizationMap>
            </amq:map>
        </amq:authorizationPlugin>
    </amq:plugins>

    <amq:systemUsage>
        <amq:systemUsage>
            <amq:memoryUsage>
                <amq:memoryUsage limit="3 gb" />
            </amq:memoryUsage>
            <amq:storeUsage>
                <amq:storeUsage limit="10 gb" />
            </amq:storeUsage>
            <amq:tempUsage>
                <amq:tempUsage limit="2 gb" />
            </amq:tempUsage>
        </amq:systemUsage>
    </amq:systemUsage>

    <amq:transportConnectors>
        <amq:transportConnector uri="tcp://${activemq.transport.host}:${activemq.transport.port}" discoveryUri="multicast://${activemq.network.multicast.address}:${activemq.network.multicast.port}?group=${activemq.network.multicast.group}" />
        <amq:transportConnector uri="vm://${activemq.broker.name}" />
    </amq:transportConnectors>

</amq:broker>

有没有人指点一下：

• 为什么 activemq 不断尝试添加已经存在的连接？
• 为什么会被记录，为什么只在警告级别？没有其他信息 (stacktrace) 记录
• 这真的是个问题吗，因为队列处理不受影响，一切运行正常？

希望收到您的来信，

金

Answer 1

查看ActiveMQ 5.10.2的源码，TransportConnection.java中出现了字符串Failed to add Connection：

try {
    broker.addConnection(context, info);
} catch (Exception e) {
    synchronized (brokerConnectionStates) {
        brokerConnectionStates.remove(info.getConnectionId());
    }
    unregisterConnectionState(info.getConnectionId());
    LOG.warn("Failed to add Connection {}", info.getConnectionId(), e);
    if (e instanceof SecurityException) {
        // close this down - in case the peer of this transport doesn't play nice
        delayedStop(2000, "Failed with SecurityException: " + e.getLocalizedMessage(), e);
    }
    throw e;
}

您的警告是由异常引起的，并且尝试的连接未注册。此时，我将检查配置文件并禁用所有不需要的服务和端口。

你能分享你的配置文件吗？