You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@activemq.apache.org by ievanliu <ie...@qq.com> on 2018/09/17 13:16:55 UTC
no consumers for the queue by a duplex network connector
activemq version: 5.14.0
jdk version: 1.7.0_67
architecture：
Broker cluster A(a 3 instances activemq cluster use jdbc lock with mysql
backend) and Broker B cluster(a 3 instances cluster use jdbc lock with
mysql backend).
Broker B config a duplex network connector to Broker A. Producer produce
messages into brokerA, consumers consume messages from broke B, bridge queue
like queueA,queueB
Also there are to more cluster connect to Broker A with a same network
connector like BrokerB
networkConnector config in Broker B (this is the compeletely config in the
real environment, no any other extra configuration):
<networkConnectors>
<networkConnector
uri="masterslave:(tcp://172.16.202.62:61616,tcp://172.16.202.65:61616,tcp://172.16.202.68:61616)"
duplex="true" name="bridge_111"></networkConnector>
</networkConnectors>
problem:
early this morning, brokeB happened a faileover event, and the master
changed. the pre-master Broker B1 lost the jdbc lock. and broker B2 in the
broke B cluster, successfuly became the master, and established the network
connection to broker A.
And then, the problem happened. producer can produce messages to Broker A,
but consumers cannot consumer messages from brokeB by the same queue.
process:
1、as the log shows, the network bridge has successfully been established。and
there are no other error logs info.
2、I found that, when consumer connect to broker B to consume queueA, in
broke A, there is no consumers for queueA.Normally, here should be a network
bridge consumer to brokeA.
3、restart the broke A and broke B，the problem happened as same
4、when the problem happened,Iuse jconsole to see the jmx info. I can see
that there certainly was no consumer of the Broker A,and there was no topic
name like "ActiveMQ.Advisory.Consumer.Queue.queueA" for the bridge queue
queueA.
5、I try to use static bridge for testing, and when static bridge was
configed fo the queueA,the problem disappeared.So it seam like the advisory
topic's problem, as I know that bridge need adbisory topic and message.
resole solution:
shutdown all the instances of broker cluster A,and broker cluster B.and then
start a instance of broker cluster A, then start a instance of broker
cluster B.
By these steps, the network bridge back to normal,and the bridge-queue has
the consumers.
this problem happened very often recently, once fail-over happened,it
happened, and I have to use the above steps to recovery the cluster.
the activemq.xml :
<beans
xmlns="http://www.springframework.org/schema/beans"
xmlns:amq="http://activemq.apache.org/schema/core"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://activemq.apache.org/schema/core
http://activemq.apache.org/schema/core/activemq-core.xsd
http://activemq.apache.org/camel/schema/spring
http://activemq.apache.org/camel/schema/spring/camel-spring.xsd">
<bean
class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
<property name="locations">
<value>file:${activemq.conf}/credentials.properties</value>
</property>
</bean>
<bean id="logQuery" class="io.fabric8.insight.log.log4j.Log4jLogQuery"
lazy-init="false" scope="singleton"
init-method="start" destroy-method="stop">
</bean>
<bean id="mysql-ds" class="org.apache.commons.dbcp2.BasicDataSource"
destroy-method="close">
<property name="driverClassName" value="com.mysql.jdbc.Driver"/>
<property name="url"
value="jdbc:mysql://*****.com:5011/activemq?relaxAutoCommit=true"/>
<property name="username" value="****"/>
<property name="password" value="****"/>
<property name="initialSize" value="10" />
<property name="maxIdle" value="10" />
<property name="minIdle" value="100" />
<property name="validationQuery" value="select 1" />
<property name="poolPreparedStatements" value="true"/>
</bean>
<broker xmlns="http://activemq.apache.org/schema/core"
brokerName="bj_172.20.32.110" dataDirectory="${activemq.data}">
<destinationPolicy>
<policyMap>
<policyEntries>
<policyEntry topic=">" >
<pendingMessageLimitStrategy>
<constantPendingMessageLimitStrategy limit="8000"/>
</pendingMessageLimitStrategy>
</policyEntry>
</policyEntries>
</policyMap>
</destinationPolicy>
<networkConnectors>
<networkConnector
uri="masterslave:(tcp://172.16.202.62:61616,tcp://172.16.202.65:61616,tcp://172.16.202.68:61616)"
duplex="true"
name="nj_to_bj_110"
>
</networkConnector>
</networkConnectors>
<managementContext>
<managementContext createConnector="true"/>
</managementContext>
<persistenceAdapter>
<jdbcPersistenceAdapter dataDirectory="activemq-data"
dataSource="#mysql-ds"/>
</persistenceAdapter>
<systemUsage>
<systemUsage>
<memoryUsage>
<memoryUsage percentOfJvmHeap="70" />
</memoryUsage>
<storeUsage>
<storeUsage limit="100 gb"/>
</storeUsage>
<tempUsage>
<tempUsage limit="50 gb"/>
</tempUsage>
</systemUsage>
</systemUsage>
<transportConnectors>
<transportConnector name="openwire"
uri="tcp://0.0.0.0:61616?maximumConnections=8000&wireFormat.maxFrameSize=2104857600"/>
<transportConnector name="amqp"
uri="amqp://0.0.0.0:5672?maximumConnections=8000&wireFormat.maxFrameSize=2104857600"/>
<transportConnector name="stomp"
uri="stomp://0.0.0.0:61613?maximumConnections=8000&wireFormat.maxFrameSize=2104857600"/>
<transportConnector name="mqtt"
uri="mqtt://0.0.0.0:1883?maximumConnections=8000&wireFormat.maxFrameSize=2104857600"/>
<transportConnector name="ws"
uri="ws://0.0.0.0:61614?maximumConnections=8000&wireFormat.maxFrameSize=2104857600"/>
</transportConnectors>
<shutdownHooks>
<bean xmlns="http://www.springframework.org/schema/beans"
class="org.apache.activemq.hooks.SpringContextHook" />
</shutdownHooks>
</broker>
<import resource="jetty.xml"/>
</beans>
--
Sent from: http://activemq.2283324.n4.nabble.com/ActiveMQ-User-f2341805.html
Re: no consumers for the queue by a duplex network connector
Posted by ievanliu <ie...@qq.com>.
I found a bug in 5.14.0 as below:
https://issues.apache.org/jira/browse/AMQ-6640
and when the problem happened,I collected the jstacek log as below which is
the same as the log in the bug's description.
and in my env, Broke A had more than 3 million pendding messages, but after
I clean up most of the pendding messages,and only left 70 thousand pendding
messages.
but the problem didn't resolve, when fail-over happen in broker cluster
B,the problem came as the same.
and the jstack log is as the same too.
jstack log:
broker 1:
"ActiveMQ Transport: tcp:///172.20.32.111:37959@61616" daemon prio=10
tid=0x00007f4e7c07a000 nid=0x282a waiting on condition [0x00007f4e52bee000]
java.lang.Thread.State: WAITING (parking)
at sun.misc.Unsafe.park(Native Method)
- parking to wait for <0x00000007657e79d0> (a
java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
at java.util.concurrent.locks.LockSupport.park(LockSupport.java:186)
at
java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2043)
at
java.util.concurrent.ArrayBlockingQueue.take(ArrayBlockingQueue.java:374)
at
org.apache.activemq.transport.FutureResponse.getResult(FutureResponse.java:48)
at
org.apache.activemq.transport.ResponseCorrelator.request(ResponseCorrelator.java:87)
at
org.apache.activemq.network.DemandForwardingBridgeSupport.addSubscription(DemandForwardingBridgeSupport.java:997)
at
org.apache.activemq.network.DemandForwardingBridgeSupport.addConsumerInfo(DemandForwardingBridgeSupport.java:1304)
at
org.apache.activemq.network.DemandForwardingBridgeSupport.serviceRemoteCommand(DemandForwardingBridgeSupport.java:752)
at
org.apache.activemq.network.DemandForwardingBridgeSupport$3.onCommand(DemandForwardingBridgeSupport.java:233)
at
org.apache.activemq.transport.ResponseCorrelator.onCommand(ResponseCorrelator.java:116)
at
org.apache.activemq.transport.MutexTransport.onCommand(MutexTransport.java:50)
at
org.apache.activemq.transport.WireFormatNegotiator.onCommand(WireFormatNegotiator.java:125)
at
org.apache.activemq.transport.AbstractInactivityMonitor.onCommand(AbstractInactivityMonitor.java:300)
at
org.apache.activemq.transport.TransportSupport.doConsume(TransportSupport.java:83)
at
org.apache.activemq.transport.tcp.TcpTransport.doRun(TcpTransport.java:233)
at
org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:215)
at java.lang.Thread.run(Thread.java:745)
"ActiveMQ VMTransport: vm://localhost#34-1" daemon prio=10
tid=0x00007f4e7db2a000 nid=0x282e runnable [0x00007f4e52fb1000]
java.lang.Thread.State: RUNNABLE
at java.net.SocketOutputStream.socketWrite0(Native Method)
at java.net.SocketOutputStream.socketWrite(SocketOutputStream.java:113)
at java.net.SocketOutputStream.write(SocketOutputStream.java:159)
at
org.apache.activemq.transport.tcp.TcpBufferedOutputStream.flush(TcpBufferedOutputStream.java:115)
at java.io.DataOutputStream.flush(DataOutputStream.java:123)
at
org.apache.activemq.transport.tcp.TcpTransport.oneway(TcpTransport.java:194)
at
org.apache.activemq.transport.AbstractInactivityMonitor.doOnewaySend(AbstractInactivityMonitor.java:334)
at
org.apache.activemq.transport.AbstractInactivityMonitor.oneway(AbstractInactivityMonitor.java:316)
at
org.apache.activemq.transport.TransportFilter.oneway(TransportFilter.java:94)
at
org.apache.activemq.transport.WireFormatNegotiator.oneway(WireFormatNegotiator.java:116)
at
org.apache.activemq.transport.MutexTransport.oneway(MutexTransport.java:68)
at
org.apache.activemq.transport.ResponseCorrelator.oneway(ResponseCorrelator.java:60)
at
org.apache.activemq.network.DemandForwardingBridgeSupport.serviceLocalCommand(DemandForwardingBridgeSupport.java:1076)
at
org.apache.activemq.network.DemandForwardingBridgeSupport$2.onCommand(DemandForwardingBridgeSupport.java:215)
at
org.apache.activemq.transport.ResponseCorrelator.onCommand(ResponseCorrelator.java:116)
at
org.apache.activemq.transport.MutexTransport.onCommand(MutexTransport.java:50)
at
org.apache.activemq.transport.vm.VMTransport.iterate(VMTransport.java:273)
at
org.apache.activemq.thread.PooledTaskRunner.runTask(PooledTaskRunner.java:133)
at
org.apache.activemq.thread.PooledTaskRunner$1.run(PooledTaskRunner.java:48)
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
at java.lang.Thread.run(Thread.java:745)
broker 2:
"ActiveMQ Transport: tcp:///172.16.202.62:61616@48506" prio=10
tid=0x00007f0914484000 nid=0x3057 waiting on condition [0x00007f082efca000]
java.lang.Thread.State: WAITING (parking)
at sun.misc.Unsafe.park(Native Method)
- parking to wait for <0x00000006c108d100> (a
java.util.concurrent.locks.ReentrantLock$NonfairSync)
at java.util.concurrent.locks.LockSupport.park(LockSupport.java:186)
at
java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:834)
at
java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireQueued(AbstractQueuedSynchronizer.java:867)
at
java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire(AbstractQueuedSynchronizer.java:1197)
at
java.util.concurrent.locks.ReentrantLock$NonfairSync.lock(ReentrantLock.java:214)
at java.util.concurrent.locks.ReentrantLock.lock(ReentrantLock.java:290)
at
org.apache.activemq.transport.MutexTransport.oneway(MutexTransport.java:66)
at
org.apache.activemq.transport.ResponseCorrelator.oneway(ResponseCorrelator.java:60)
at
org.apache.activemq.broker.TransportConnection.dispatch(TransportConnection.java:1506)
at
org.apache.activemq.broker.TransportConnection.processDispatch(TransportConnection.java:978)
at
org.apache.activemq.broker.TransportConnection.dispatchSync(TransportConnection.java:934)
at
org.apache.activemq.broker.TransportConnection$1.onCommand(TransportConnection.java:199)
at
org.apache.activemq.transport.ResponseCorrelator.onCommand(ResponseCorrelator.java:116)
at
org.apache.activemq.transport.MutexTransport.onCommand(MutexTransport.java:50)
at
org.apache.activemq.transport.vm.VMTransport.doDispatch(VMTransport.java:164)
at
org.apache.activemq.transport.vm.VMTransport.dispatch(VMTransport.java:156)
- locked <0x00000006c108d2b8> (a java.util.concurrent.atomic.AtomicBoolean)
at
org.apache.activemq.transport.vm.VMTransport.oneway(VMTransport.java:133)
at
org.apache.activemq.transport.MutexTransport.oneway(MutexTransport.java:68)
at
org.apache.activemq.transport.ResponseCorrelator.asyncRequest(ResponseCorrelator.java:81)
at
org.apache.activemq.transport.ResponseCorrelator.request(ResponseCorrelator.java:86)
at
org.apache.activemq.network.DemandForwardingBridgeSupport.addSubscription(DemandForwardingBridgeSupport.java:997)
at
org.apache.activemq.network.DemandForwardingBridgeSupport.addConsumerInfo(DemandForwardingBridgeSupport.java:1304)
at
org.apache.activemq.network.DemandForwardingBridgeSupport.serviceRemoteConsumerAdvisory(DemandForwardingBridgeSupport.java:835)
- locked <0x00000006c0203078> (a java.net.URI)
at
org.apache.activemq.network.DemandForwardingBridgeSupport.serviceRemoteCommand(DemandForwardingBridgeSupport.java:682)
at
org.apache.activemq.network.DemandForwardingBridgeSupport$3.onCommand(DemandForwardingBridgeSupport.java:233)
at
org.apache.activemq.transport.ResponseCorrelator.onCommand(ResponseCorrelator.java:116)
at
org.apache.activemq.transport.MutexTransport.onCommand(MutexTransport.java:50)
at
org.apache.activemq.transport.failover.FailoverTransport.processCommand(FailoverTransport.java:208)
at
org.apache.activemq.transport.failover.FailoverTransport.access$1100(FailoverTransport.java:73)
at
org.apache.activemq.transport.failover.FailoverTransport$3.onCommand(FailoverTransport.java:217)
at
org.apache.activemq.transport.WireFormatNegotiator.onCommand(WireFormatNegotiator.java:125)
at
org.apache.activemq.transport.AbstractInactivityMonitor.onCommand(AbstractInactivityMonitor.java:300)
at
org.apache.activemq.transport.TransportSupport.doConsume(TransportSupport.java:83)
at
org.apache.activemq.transport.tcp.TcpTransport.doRun(TcpTransport.java:233)
at
org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:215)
at java.lang.Thread.run(Thread.java:745)
"ActiveMQ BrokerService[bj_172.20.32.110] Task-57" daemon prio=10
tid=0x00007f08b8077000 nid=0x2ffb runnable [0x00007f0835f63000]
java.lang.Thread.State: RUNNABLE
at java.net.SocketOutputStream.socketWrite0(Native Method)
at java.net.SocketOutputStream.socketWrite(SocketOutputStream.java:113)
at java.net.SocketOutputStream.write(SocketOutputStream.java:159)
at
org.apache.activemq.transport.tcp.TcpBufferedOutputStream.flush(TcpBufferedOutputStream.java:115)
at java.io.DataOutputStream.flush(DataOutputStream.java:123)
at
org.apache.activemq.transport.tcp.TcpTransport.oneway(TcpTransport.java:194)
at
org.apache.activemq.transport.AbstractInactivityMonitor.doOnewaySend(AbstractInactivityMonitor.java:334)
at
org.apache.activemq.transport.AbstractInactivityMonitor.oneway(AbstractInactivityMonitor.java:316)
at
org.apache.activemq.transport.TransportFilter.oneway(TransportFilter.java:94)
at
org.apache.activemq.transport.WireFormatNegotiator.oneway(WireFormatNegotiator.java:116)
at
org.apache.activemq.transport.failover.FailoverTransport.oneway(FailoverTransport.java:668)
- locked <0x00000006c108de58> (a java.lang.Object)
at
org.apache.activemq.transport.MutexTransport.oneway(MutexTransport.java:68)
at
org.apache.activemq.transport.ResponseCorrelator.oneway(ResponseCorrelator.java:60)
at
org.apache.activemq.network.DemandForwardingBridgeSupport.serviceLocalCommand(DemandForwardingBridgeSupport.java:1076)
at
org.apache.activemq.network.DemandForwardingBridgeSupport$2.onCommand(DemandForwardingBridgeSupport.java:215)
at
org.apache.activemq.transport.ResponseCorrelator.onCommand(ResponseCorrelator.java:116)
at
org.apache.activemq.transport.MutexTransport.onCommand(MutexTransport.java:50)
at
org.apache.activemq.transport.vm.VMTransport.doDispatch(VMTransport.java:164)
at
org.apache.activemq.transport.vm.VMTransport.dispatch(VMTransport.java:156)
- locked <0x00000006c108d2d8> (a java.util.concurrent.atomic.AtomicBoolean)
at
org.apache.activemq.transport.vm.VMTransport.oneway(VMTransport.java:133)
at
org.apache.activemq.transport.MutexTransport.oneway(MutexTransport.java:68)
at
org.apache.activemq.transport.ResponseCorrelator.oneway(ResponseCorrelator.java:60)
at
org.apache.activemq.broker.TransportConnection.dispatch(TransportConnection.java:1506)
at
org.apache.activemq.broker.TransportConnection.processDispatch(TransportConnection.java:978)
at
org.apache.activemq.broker.TransportConnection.iterate(TransportConnection.java:1024)
at
org.apache.activemq.thread.PooledTaskRunner.runTask(PooledTaskRunner.java:133)
at
org.apache.activemq.thread.PooledTaskRunner$1.run(PooledTaskRunner.java:48)
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
at java.lang.Thread.run(Thread.java:745)
--
Sent from: http://activemq.2283324.n4.nabble.com/ActiveMQ-User-f2341805.html