You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@activemq.apache.org by David Delbecq <da...@oma.be> on 2010/08/03 10:56:11 UTC
QueueReceiver.receive(timeout) never return messages
Hello,
we are having fiability troubles with activeMQ (5.2). To get around
those, we use in consumer receive(timeout) with a timeout of 60 seconds
to detect early problems with broker. We assumed if we called receive()
on a closed connection (broker side closed) we sould somehow get an
exception. However, when activeMQ server is shutdown for any reason,
the receiver never detect this and the receive call never return any
message. We suspect the timeout occurs, but we would expect some kind of
exception telling us we need to reconnect.
Here is consumer code:
while (!stopNow) {
Message m = receiver.receive(60000);//wait 60 then next
loop
if (m == null) {
continue; // timeout?
}
if (m instanceof MapMessage) { // The event queue
should contain only map messages !
processEventMessage((MapMessage) m);
Here is the threaddump of consumer. There are messages in that queue
waiting since yesterday, but the consumer never received any of those!
This is very problematic as, for now, we must restart every consumer
after restarting the broker. We would expect the consumer to be able to
detect this and reconnect. What's the procedure to follow for this? We
could disconnect / reconnect every minutes to be sure, but that would
mean some kind of additional load on the broker.
INFO | jvm 1 | 2010/08/03 08:35:35 | "ActiveMQ Connection Worker:
tcp://localhost/127.0.0.1:61616" daemon prio=10 tid=0x00007f1cc1d7f000
nid=0x5d3a waiting on condition [0x0000000041f7b000]
INFO | jvm 1 | 2010/08/03 08:35:35 | java.lang.Thread.State:
WAITING (parking)
INFO | jvm 1 | 2010/08/03 08:35:35 | at
sun.misc.Unsafe.park(Native Method)
INFO | jvm 1 | 2010/08/03 08:35:35 | - parking to wait for
<0x00007f1cc8412618> (a
java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
INFO | jvm 1 | 2010/08/03 08:35:35 | at
java.util.concurrent.locks.LockSupport.park(LockSupport.java:158)
INFO | jvm 1 | 2010/08/03 08:35:35 | at
java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:1925)
INFO | jvm 1 | 2010/08/03 08:35:35 | at
java.util.concurrent.LinkedBlockingQueue.take(LinkedBlockingQueue.java:358)
INFO | jvm 1 | 2010/08/03 08:35:35 | at
java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:947)
INFO | jvm 1 | 2010/08/03 08:35:35 | at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:907)
INFO | jvm 1 | 2010/08/03 08:35:35 | at
java.lang.Thread.run(Thread.java:619)
INFO | jvm 1 | 2010/08/03 08:35:35 | "WrapperJarAppMain" prio=10
tid=0x00007f1cc22b8000 nid=0x5f4d in Object.wait() [0x0000000040af6000]
INFO | jvm 1 | 2010/08/03 08:35:35 | java.lang.Thread.State:
TIMED_WAITING (on object monitor)
INFO | jvm 1 | 2010/08/03 08:35:35 | at
java.lang.Object.wait(Native Method)
INFO | jvm 1 | 2010/08/03 08:35:35 | - waiting on
<0x00007f1cc8413280> (a java.lang.Object)
INFO | jvm 1 | 2010/08/03 08:35:35 | at
org.apache.activemq.MessageDispatchChannel.dequeue(MessageDispatchChannel.java:77)
INFO | jvm 1 | 2010/08/03 08:35:35 | - locked
<0x00007f1cc8413280> (a java.lang.Object)
INFO | jvm 1 | 2010/08/03 08:35:35 | at
org.apache.activemq.ActiveMQMessageConsumer.dequeue(ActiveMQMessageConsumer.java:412)
INFO | jvm 1 | 2010/08/03 08:35:35 | at
org.apache.activemq.ActiveMQMessageConsumer.receive(ActiveMQMessageConsumer.java:531)
INFO | jvm 1 | 2010/08/03 08:35:35 | at
be.meteo.shark.client.emailer.JmsEmailer.processMessages(JmsEmailer.java:178)
INFO | jvm 1 | 2010/08/03 08:35:35 | at
be.meteo.shark.client.emailer.JmsEmailer.main(JmsEmailer.java:409)
INFO | jvm 1 | 2010/08/03 08:35:35 | at
sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
INFO | jvm 1 | 2010/08/03 08:35:35 | at
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
INFO | jvm 1 | 2010/08/03 08:35:35 | at
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
INFO | jvm 1 | 2010/08/03 08:35:35 | at
java.lang.reflect.Method.invoke(Method.java:597)
INFO | jvm 1 | 2010/08/03 08:35:35 | at
org.tanukisoftware.wrapper.WrapperJarApp.run(WrapperJarApp.java:352)
INFO | jvm 1 | 2010/08/03 08:35:35 | at
java.lang.Thread.run(Thread.java:619)
--
David Delbecq
ICT
Institut Royal Météorologique
Ext:557
Help with Activemq Spring support
Posted by qin ding <qi...@yahoo.com>.
ActiveMQ 5.3.2
Spring 2.0
I have a simple producer:
public class AdminProducer {
private static final Log log = LogFactory.getLog(AdminProducer.class);
private JmsTemplate jmsTemplate;
private Destination destination;
public void publish(final AdminInfo adminInfo) throws JMSException{
log.debug("in publish..."
jmsTemplate.send(destination, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
TextMessage message =
session.createTextMessage(adminInfo.toXml());
return message;
}
});
}
public JmsTemplate getJmsTemplate() {
return jmsTemplate;
}
public void setJmsTemplate(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
public Destination getDestination() {
return destination;
}
public void setDestination(Destination destination) {
this.destination = destination;
}
}
applicationContext.xml:
<!-- admin info producer -->
<bean id="adminProducer" class="com.mgt.lottery.poker.server.AdminProducer">
<property name="jmsTemplate" ref="jmsTemplate" />
<property name="destination" ref="adminTopic" />
</bean>
<!-- JMS ConnectionFactory to use -->
<bean id="jmsFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://localhost:61616/>
</bean>
<!-- Spring JMS Template -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory">
<bean
class="org.springframework.jms.connection.SingleConnectionFactory">
<property name="targetConnectionFactory" ref="jmsFactory" />
</bean>
</property>
</bean>
<!-- admin destination topic -->
<bean id="adminTopic" class="org.apache.activemq.command.ActiveMQTopic"
autowire="constructor">
<constructor-arg value="mgt.lottery.poker.admin.topic" />
</bean>
I got the error:
Caused by: org.springframework.beans.TypeMismatchException: Failed to convert
property value of type [$Proxy5 implementing
org.springframework.jms.core.JmsOpera
tions,org.springframework.beans.factory.InitializingBean,org.springframework.aop.SpringProxy,org.springframework.aop.framework.Advised]
to required type [org.sp
ringframework.jms.core.JmsTemplate] for property 'jmsTemplate'; nested
exception is java.lang.IllegalArgumentException: Cannot convert value of type
[$Proxy5 im
plementing
org.springframework.jms.core.JmsOperations,org.springframework.beans.factory.InitializingBean,org.springframework.aop.SpringProxy,org.springframework
.aop.framework.Advised] to required type
[org.springframework.jms.core.JmsTemplate] for property 'jmsTemplate': no
matching editors or conversion strategy found
at
org.springframework.beans.BeanWrapperImpl.convertForProperty(BeanWrapperImpl.java:391)
at
org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.convertForProperty(AbstractAutowireCapableBeanFactory.java:1289)
at
org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.applyPropertyValues(AbstractAutowireCapableBeanFactory.java:1250)
at
org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.populateBean(AbstractAutowireCapableBeanFactory.java:1010)
at
org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:472)
... 27 more
Caused by: java.lang.IllegalArgumentException: Cannot convert value of type
[$Proxy5 implementing
org.springframework.jms.core.JmsOperations,org.springframework
.beans.factory.InitializingBean,org.springframework.aop.SpringProxy,org.springframework.aop.framework.Advised]
to required type [org.springframework.jms.core.Jm
sTemplate] for property 'jmsTemplate': no matching editors or conversion
strategy found
What should I do to fix this? Please help. Thanks.
QD
Re: QueueReceiver.receive(timeout) never return messages
Posted by Gary Tully <ga...@gmail.com>.
In the regular tcp: case, the receive sends an async command to the
broker and then blocks on the internal consumer queue, if that async
send were to fail you would need an exception listener to capture it.
The failover transort traps all exceptions on async/oneway requests so
it will notice by default.
On 3 August 2010 16:08, David Delbecq <da...@oma.be> wrote:
> Hello,
>
> thanks, its seems to work around our problem, we will keep an eye on it.
> Are there any reason while failover: connection is capable of detecting the
> network problem and reconnecting while the tcp: just "wait"? Shouldn't it at
> least throw an exception when link is down?
>
>
> Le 03/08/10 16:52, Dejan Bosanac a écrit :
>>
>> Hi David,
>>
>> failover transport can be used for just one broker as well and it is
>> used to solve these kind of problems. So
>>
>> failover:(tcp://localhost:61616)
>>
>> should work just fine. If the broker is down or there are network
>> problems, the transport will try to reestablish the connection.
>>
>>
>>
>> Cheers
>> --
>> Dejan Bosanac - http://twitter.com/dejanb
>>
>> Open Source Integration - http://fusesource.com/
>> ActiveMQ in Action - http://www.manning.com/snyder/
>> Blog - http://www.nighttale.net
>>
>>
>>
>> On Tue, Aug 3, 2010 at 4:46 PM, David Delbecq<da...@oma.be>
>> wrote:
>>
>>>
>>> Thanks for reply
>>>
>>> we do not use failover, because there is only one broker. It's not
>>> critical
>>> enough to justify two or more brokers. We just want receive() to fail in
>>> a
>>> way or other when the other ends (broker) dies for some reason, so we can
>>> take appropriate measures. This is just illogical for receive(timeout) to
>>> just return null when the underlying connection does not exist anymore.
>>>
>>>
>>> Le 03/08/10 16:08, Dejan Bosanac a écrit :
>>>
>>>>
>>>> Hi David,
>>>>
>>>> did you try using failover transport
>>>>
>>>> http://activemq.apache.org/failover-transport-reference.html
>>>>
>>>> It should take care of detecting network problems and reconnecting.
>>>> Then you should just use receive() or message listener
>>>>
>>>> Cheers
>>>> --
>>>> Dejan Bosanac - http://twitter.com/dejanb
>>>>
>>>> Open Source Integration - http://fusesource.com/
>>>> ActiveMQ in Action - http://www.manning.com/snyder/
>>>> Blog - http://www.nighttale.net
>>>>
>>>>
>>>>
>>>> On Tue, Aug 3, 2010 at 10:56 AM, David Delbecq<da...@oma.be>
>>>> wrote:
>>>>
>>>>
>>>>>
>>>>> Hello,
>>>>>
>>>>> we are having fiability troubles with activeMQ (5.2). To get around
>>>>> those,
>>>>> we use in consumer receive(timeout) with a timeout of 60 seconds to
>>>>> detect
>>>>> early problems with broker. We assumed if we called receive() on a
>>>>> closed
>>>>> connection (broker side closed) we sould somehow get an exception.
>>>>> However,
>>>>> when activeMQ server is shutdown for any reason, the receiver never
>>>>> detect
>>>>> this and the receive call never return any message. We suspect the
>>>>> timeout
>>>>> occurs, but we would expect some kind of exception telling us we need
>>>>> to
>>>>> reconnect.
>>>>>
>>>>>
>>>>> Here is consumer code:
>>>>> while (!stopNow) {
>>>>> Message m = receiver.receive(60000);//wait 60 then next
>>>>> loop
>>>>> if (m == null) {
>>>>> continue; // timeout?
>>>>> }
>>>>> if (m instanceof MapMessage) { // The event queue should
>>>>> contain only map messages !
>>>>> processEventMessage((MapMessage) m);
>>>>>
>>>>>
>>>>> Here is the threaddump of consumer. There are messages in that queue
>>>>> waiting
>>>>> since yesterday, but the consumer never received any of those! This is
>>>>> very
>>>>> problematic as, for now, we must restart every consumer after
>>>>> restarting
>>>>> the
>>>>> broker. We would expect the consumer to be able to detect this and
>>>>> reconnect. What's the procedure to follow for this? We could disconnect
>>>>> /
>>>>> reconnect every minutes to be sure, but that would mean some kind of
>>>>> additional load on the broker.
>>>>>
>>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | "ActiveMQ Connection Worker:
>>>>> tcp://localhost/127.0.0.1:61616" daemon prio=10 tid=0x00007f1cc1d7f000
>>>>> nid=0x5d3a waiting on condition [0x0000000041f7b000]
>>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | java.lang.Thread.State:
>>>>> WAITING
>>>>> (parking)
>>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>>> sun.misc.Unsafe.park(Native
>>>>> Method)
>>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | - parking to wait for
>>>>> <0x00007f1cc8412618> (a
>>>>> java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
>>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>>> java.util.concurrent.locks.LockSupport.park(LockSupport.java:158)
>>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>>>
>>>>>
>>>>> java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:1925)
>>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>>>
>>>>>
>>>>> java.util.concurrent.LinkedBlockingQueue.take(LinkedBlockingQueue.java:358)
>>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>>>
>>>>>
>>>>> java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:947)
>>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>>>
>>>>>
>>>>> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:907)
>>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>>> java.lang.Thread.run(Thread.java:619)
>>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | "WrapperJarAppMain" prio=10
>>>>> tid=0x00007f1cc22b8000 nid=0x5f4d in Object.wait() [0x0000000040af6000]
>>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | java.lang.Thread.State:
>>>>> TIMED_WAITING (on object monitor)
>>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>>> java.lang.Object.wait(Native Method)
>>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | - waiting on
>>>>> <0x00007f1cc8413280> (a java.lang.Object)
>>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>>>
>>>>>
>>>>> org.apache.activemq.MessageDispatchChannel.dequeue(MessageDispatchChannel.java:77)
>>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | -
>>>>> locked<0x00007f1cc8413280>
>>>>> (a java.lang.Object)
>>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>>>
>>>>>
>>>>> org.apache.activemq.ActiveMQMessageConsumer.dequeue(ActiveMQMessageConsumer.java:412)
>>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>>>
>>>>>
>>>>> org.apache.activemq.ActiveMQMessageConsumer.receive(ActiveMQMessageConsumer.java:531)
>>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>>>
>>>>>
>>>>> be.meteo.shark.client.emailer.JmsEmailer.processMessages(JmsEmailer.java:178)
>>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>>> be.meteo.shark.client.emailer.JmsEmailer.main(JmsEmailer.java:409)
>>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>>> sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
>>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>>>
>>>>>
>>>>> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
>>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>>>
>>>>>
>>>>> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
>>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>>> java.lang.reflect.Method.invoke(Method.java:597)
>>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>>> org.tanukisoftware.wrapper.WrapperJarApp.run(WrapperJarApp.java:352)
>>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>>> java.lang.Thread.run(Thread.java:619)
>>>>>
>>>>> --
>>>>> David Delbecq
>>>>> ICT
>>>>> Institut Royal Météorologique
>>>>> Ext:557
>>>>>
>>>>>
>>>>>
>>>>>
>>>>>
>>>
>>> --
>>> David Delbecq
>>> ICT
>>> Institut Royal Météorologique
>>> Ext:557
>>>
>>>
>>>
>>>
>
>
> --
> David Delbecq
> ICT
> Institut Royal Météorologique
> Ext:557
>
>
>
--
http://blog.garytully.com
Open Source Integration
http://fusesource.com
Re: QueueReceiver.receive(timeout) never return messages
Posted by David Delbecq <da...@oma.be>.
Hello,
thanks, its seems to work around our problem, we will keep an eye on
it. Are there any reason while failover: connection is capable of
detecting the network problem and reconnecting while the tcp: just
"wait"? Shouldn't it at least throw an exception when link is down?
Le 03/08/10 16:52, Dejan Bosanac a écrit :
> Hi David,
>
> failover transport can be used for just one broker as well and it is
> used to solve these kind of problems. So
>
> failover:(tcp://localhost:61616)
>
> should work just fine. If the broker is down or there are network
> problems, the transport will try to reestablish the connection.
>
>
>
> Cheers
> --
> Dejan Bosanac - http://twitter.com/dejanb
>
> Open Source Integration - http://fusesource.com/
> ActiveMQ in Action - http://www.manning.com/snyder/
> Blog - http://www.nighttale.net
>
>
>
> On Tue, Aug 3, 2010 at 4:46 PM, David Delbecq<da...@oma.be> wrote:
>
>> Thanks for reply
>>
>> we do not use failover, because there is only one broker. It's not critical
>> enough to justify two or more brokers. We just want receive() to fail in a
>> way or other when the other ends (broker) dies for some reason, so we can
>> take appropriate measures. This is just illogical for receive(timeout) to
>> just return null when the underlying connection does not exist anymore.
>>
>>
>> Le 03/08/10 16:08, Dejan Bosanac a écrit :
>>
>>> Hi David,
>>>
>>> did you try using failover transport
>>>
>>> http://activemq.apache.org/failover-transport-reference.html
>>>
>>> It should take care of detecting network problems and reconnecting.
>>> Then you should just use receive() or message listener
>>>
>>> Cheers
>>> --
>>> Dejan Bosanac - http://twitter.com/dejanb
>>>
>>> Open Source Integration - http://fusesource.com/
>>> ActiveMQ in Action - http://www.manning.com/snyder/
>>> Blog - http://www.nighttale.net
>>>
>>>
>>>
>>> On Tue, Aug 3, 2010 at 10:56 AM, David Delbecq<da...@oma.be>
>>> wrote:
>>>
>>>
>>>> Hello,
>>>>
>>>> we are having fiability troubles with activeMQ (5.2). To get around
>>>> those,
>>>> we use in consumer receive(timeout) with a timeout of 60 seconds to
>>>> detect
>>>> early problems with broker. We assumed if we called receive() on a
>>>> closed
>>>> connection (broker side closed) we sould somehow get an exception.
>>>> However,
>>>> when activeMQ server is shutdown for any reason, the receiver never
>>>> detect
>>>> this and the receive call never return any message. We suspect the
>>>> timeout
>>>> occurs, but we would expect some kind of exception telling us we need to
>>>> reconnect.
>>>>
>>>>
>>>> Here is consumer code:
>>>> while (!stopNow) {
>>>> Message m = receiver.receive(60000);//wait 60 then next
>>>> loop
>>>> if (m == null) {
>>>> continue; // timeout?
>>>> }
>>>> if (m instanceof MapMessage) { // The event queue should
>>>> contain only map messages !
>>>> processEventMessage((MapMessage) m);
>>>>
>>>>
>>>> Here is the threaddump of consumer. There are messages in that queue
>>>> waiting
>>>> since yesterday, but the consumer never received any of those! This is
>>>> very
>>>> problematic as, for now, we must restart every consumer after restarting
>>>> the
>>>> broker. We would expect the consumer to be able to detect this and
>>>> reconnect. What's the procedure to follow for this? We could disconnect
>>>> /
>>>> reconnect every minutes to be sure, but that would mean some kind of
>>>> additional load on the broker.
>>>>
>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | "ActiveMQ Connection Worker:
>>>> tcp://localhost/127.0.0.1:61616" daemon prio=10 tid=0x00007f1cc1d7f000
>>>> nid=0x5d3a waiting on condition [0x0000000041f7b000]
>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | java.lang.Thread.State:
>>>> WAITING
>>>> (parking)
>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>> sun.misc.Unsafe.park(Native
>>>> Method)
>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | - parking to wait for
>>>> <0x00007f1cc8412618> (a
>>>> java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>> java.util.concurrent.locks.LockSupport.park(LockSupport.java:158)
>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>>
>>>> java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:1925)
>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>>
>>>> java.util.concurrent.LinkedBlockingQueue.take(LinkedBlockingQueue.java:358)
>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>>
>>>> java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:947)
>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>>
>>>> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:907)
>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>> java.lang.Thread.run(Thread.java:619)
>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | "WrapperJarAppMain" prio=10
>>>> tid=0x00007f1cc22b8000 nid=0x5f4d in Object.wait() [0x0000000040af6000]
>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | java.lang.Thread.State:
>>>> TIMED_WAITING (on object monitor)
>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>> java.lang.Object.wait(Native Method)
>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | - waiting on
>>>> <0x00007f1cc8413280> (a java.lang.Object)
>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>>
>>>> org.apache.activemq.MessageDispatchChannel.dequeue(MessageDispatchChannel.java:77)
>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | -
>>>> locked<0x00007f1cc8413280>
>>>> (a java.lang.Object)
>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>>
>>>> org.apache.activemq.ActiveMQMessageConsumer.dequeue(ActiveMQMessageConsumer.java:412)
>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>>
>>>> org.apache.activemq.ActiveMQMessageConsumer.receive(ActiveMQMessageConsumer.java:531)
>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>>
>>>> be.meteo.shark.client.emailer.JmsEmailer.processMessages(JmsEmailer.java:178)
>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>> be.meteo.shark.client.emailer.JmsEmailer.main(JmsEmailer.java:409)
>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>> sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>>
>>>> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>>
>>>> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>> java.lang.reflect.Method.invoke(Method.java:597)
>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>> org.tanukisoftware.wrapper.WrapperJarApp.run(WrapperJarApp.java:352)
>>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>> java.lang.Thread.run(Thread.java:619)
>>>>
>>>> --
>>>> David Delbecq
>>>> ICT
>>>> Institut Royal Météorologique
>>>> Ext:557
>>>>
>>>>
>>>>
>>>>
>>>>
>>
>> --
>> David Delbecq
>> ICT
>> Institut Royal Météorologique
>> Ext:557
>>
>>
>>
>>
--
David Delbecq
ICT
Institut Royal Météorologique
Ext:557
Re: QueueReceiver.receive(timeout) never return messages
Posted by Dejan Bosanac <de...@nighttale.net>.
Hi David,
failover transport can be used for just one broker as well and it is
used to solve these kind of problems. So
failover:(tcp://localhost:61616)
should work just fine. If the broker is down or there are network
problems, the transport will try to reestablish the connection.
Cheers
--
Dejan Bosanac - http://twitter.com/dejanb
Open Source Integration - http://fusesource.com/
ActiveMQ in Action - http://www.manning.com/snyder/
Blog - http://www.nighttale.net
On Tue, Aug 3, 2010 at 4:46 PM, David Delbecq <da...@oma.be> wrote:
> Thanks for reply
>
> we do not use failover, because there is only one broker. It's not critical
> enough to justify two or more brokers. We just want receive() to fail in a
> way or other when the other ends (broker) dies for some reason, so we can
> take appropriate measures. This is just illogical for receive(timeout) to
> just return null when the underlying connection does not exist anymore.
>
>
> Le 03/08/10 16:08, Dejan Bosanac a écrit :
>>
>> Hi David,
>>
>> did you try using failover transport
>>
>> http://activemq.apache.org/failover-transport-reference.html
>>
>> It should take care of detecting network problems and reconnecting.
>> Then you should just use receive() or message listener
>>
>> Cheers
>> --
>> Dejan Bosanac - http://twitter.com/dejanb
>>
>> Open Source Integration - http://fusesource.com/
>> ActiveMQ in Action - http://www.manning.com/snyder/
>> Blog - http://www.nighttale.net
>>
>>
>>
>> On Tue, Aug 3, 2010 at 10:56 AM, David Delbecq<da...@oma.be>
>> wrote:
>>
>>>
>>> Hello,
>>>
>>> we are having fiability troubles with activeMQ (5.2). To get around
>>> those,
>>> we use in consumer receive(timeout) with a timeout of 60 seconds to
>>> detect
>>> early problems with broker. We assumed if we called receive() on a
>>> closed
>>> connection (broker side closed) we sould somehow get an exception.
>>> However,
>>> when activeMQ server is shutdown for any reason, the receiver never
>>> detect
>>> this and the receive call never return any message. We suspect the
>>> timeout
>>> occurs, but we would expect some kind of exception telling us we need to
>>> reconnect.
>>>
>>>
>>> Here is consumer code:
>>> while (!stopNow) {
>>> Message m = receiver.receive(60000);//wait 60 then next
>>> loop
>>> if (m == null) {
>>> continue; // timeout?
>>> }
>>> if (m instanceof MapMessage) { // The event queue should
>>> contain only map messages !
>>> processEventMessage((MapMessage) m);
>>>
>>>
>>> Here is the threaddump of consumer. There are messages in that queue
>>> waiting
>>> since yesterday, but the consumer never received any of those! This is
>>> very
>>> problematic as, for now, we must restart every consumer after restarting
>>> the
>>> broker. We would expect the consumer to be able to detect this and
>>> reconnect. What's the procedure to follow for this? We could disconnect
>>> /
>>> reconnect every minutes to be sure, but that would mean some kind of
>>> additional load on the broker.
>>>
>>> INFO | jvm 1 | 2010/08/03 08:35:35 | "ActiveMQ Connection Worker:
>>> tcp://localhost/127.0.0.1:61616" daemon prio=10 tid=0x00007f1cc1d7f000
>>> nid=0x5d3a waiting on condition [0x0000000041f7b000]
>>> INFO | jvm 1 | 2010/08/03 08:35:35 | java.lang.Thread.State:
>>> WAITING
>>> (parking)
>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>> sun.misc.Unsafe.park(Native
>>> Method)
>>> INFO | jvm 1 | 2010/08/03 08:35:35 | - parking to wait for
>>> <0x00007f1cc8412618> (a
>>> java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>> java.util.concurrent.locks.LockSupport.park(LockSupport.java:158)
>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>
>>> java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:1925)
>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>
>>> java.util.concurrent.LinkedBlockingQueue.take(LinkedBlockingQueue.java:358)
>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>
>>> java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:947)
>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>
>>> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:907)
>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>> java.lang.Thread.run(Thread.java:619)
>>> INFO | jvm 1 | 2010/08/03 08:35:35 | "WrapperJarAppMain" prio=10
>>> tid=0x00007f1cc22b8000 nid=0x5f4d in Object.wait() [0x0000000040af6000]
>>> INFO | jvm 1 | 2010/08/03 08:35:35 | java.lang.Thread.State:
>>> TIMED_WAITING (on object monitor)
>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>> java.lang.Object.wait(Native Method)
>>> INFO | jvm 1 | 2010/08/03 08:35:35 | - waiting on
>>> <0x00007f1cc8413280> (a java.lang.Object)
>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>
>>> org.apache.activemq.MessageDispatchChannel.dequeue(MessageDispatchChannel.java:77)
>>> INFO | jvm 1 | 2010/08/03 08:35:35 | -
>>> locked<0x00007f1cc8413280>
>>> (a java.lang.Object)
>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>
>>> org.apache.activemq.ActiveMQMessageConsumer.dequeue(ActiveMQMessageConsumer.java:412)
>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>
>>> org.apache.activemq.ActiveMQMessageConsumer.receive(ActiveMQMessageConsumer.java:531)
>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>
>>> be.meteo.shark.client.emailer.JmsEmailer.processMessages(JmsEmailer.java:178)
>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>> be.meteo.shark.client.emailer.JmsEmailer.main(JmsEmailer.java:409)
>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>> sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>
>>> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>>
>>> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>> java.lang.reflect.Method.invoke(Method.java:597)
>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>> org.tanukisoftware.wrapper.WrapperJarApp.run(WrapperJarApp.java:352)
>>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>>> java.lang.Thread.run(Thread.java:619)
>>>
>>> --
>>> David Delbecq
>>> ICT
>>> Institut Royal Météorologique
>>> Ext:557
>>>
>>>
>>>
>>>
>
>
> --
> David Delbecq
> ICT
> Institut Royal Météorologique
> Ext:557
>
>
>
Re: QueueReceiver.receive(timeout) never return messages
Posted by David Delbecq <da...@oma.be>.
Thanks for reply
we do not use failover, because there is only one broker. It's not
critical enough to justify two or more brokers. We just want receive()
to fail in a way or other when the other ends (broker) dies for some
reason, so we can take appropriate measures. This is just illogical for
receive(timeout) to just return null when the underlying connection does
not exist anymore.
Le 03/08/10 16:08, Dejan Bosanac a écrit :
> Hi David,
>
> did you try using failover transport
>
> http://activemq.apache.org/failover-transport-reference.html
>
> It should take care of detecting network problems and reconnecting.
> Then you should just use receive() or message listener
>
> Cheers
> --
> Dejan Bosanac - http://twitter.com/dejanb
>
> Open Source Integration - http://fusesource.com/
> ActiveMQ in Action - http://www.manning.com/snyder/
> Blog - http://www.nighttale.net
>
>
>
> On Tue, Aug 3, 2010 at 10:56 AM, David Delbecq<da...@oma.be> wrote:
>
>> Hello,
>>
>> we are having fiability troubles with activeMQ (5.2). To get around those,
>> we use in consumer receive(timeout) with a timeout of 60 seconds to detect
>> early problems with broker. We assumed if we called receive() on a closed
>> connection (broker side closed) we sould somehow get an exception. However,
>> when activeMQ server is shutdown for any reason, the receiver never detect
>> this and the receive call never return any message. We suspect the timeout
>> occurs, but we would expect some kind of exception telling us we need to
>> reconnect.
>>
>>
>> Here is consumer code:
>> while (!stopNow) {
>> Message m = receiver.receive(60000);//wait 60 then next loop
>> if (m == null) {
>> continue; // timeout?
>> }
>> if (m instanceof MapMessage) { // The event queue should
>> contain only map messages !
>> processEventMessage((MapMessage) m);
>>
>>
>> Here is the threaddump of consumer. There are messages in that queue waiting
>> since yesterday, but the consumer never received any of those! This is very
>> problematic as, for now, we must restart every consumer after restarting the
>> broker. We would expect the consumer to be able to detect this and
>> reconnect. What's the procedure to follow for this? We could disconnect /
>> reconnect every minutes to be sure, but that would mean some kind of
>> additional load on the broker.
>>
>> INFO | jvm 1 | 2010/08/03 08:35:35 | "ActiveMQ Connection Worker:
>> tcp://localhost/127.0.0.1:61616" daemon prio=10 tid=0x00007f1cc1d7f000
>> nid=0x5d3a waiting on condition [0x0000000041f7b000]
>> INFO | jvm 1 | 2010/08/03 08:35:35 | java.lang.Thread.State: WAITING
>> (parking)
>> INFO | jvm 1 | 2010/08/03 08:35:35 | at sun.misc.Unsafe.park(Native
>> Method)
>> INFO | jvm 1 | 2010/08/03 08:35:35 | - parking to wait for
>> <0x00007f1cc8412618> (a
>> java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>> java.util.concurrent.locks.LockSupport.park(LockSupport.java:158)
>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>> java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:1925)
>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>> java.util.concurrent.LinkedBlockingQueue.take(LinkedBlockingQueue.java:358)
>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>> java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:947)
>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:907)
>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>> java.lang.Thread.run(Thread.java:619)
>> INFO | jvm 1 | 2010/08/03 08:35:35 | "WrapperJarAppMain" prio=10
>> tid=0x00007f1cc22b8000 nid=0x5f4d in Object.wait() [0x0000000040af6000]
>> INFO | jvm 1 | 2010/08/03 08:35:35 | java.lang.Thread.State:
>> TIMED_WAITING (on object monitor)
>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>> java.lang.Object.wait(Native Method)
>> INFO | jvm 1 | 2010/08/03 08:35:35 | - waiting on
>> <0x00007f1cc8413280> (a java.lang.Object)
>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>> org.apache.activemq.MessageDispatchChannel.dequeue(MessageDispatchChannel.java:77)
>> INFO | jvm 1 | 2010/08/03 08:35:35 | - locked<0x00007f1cc8413280>
>> (a java.lang.Object)
>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>> org.apache.activemq.ActiveMQMessageConsumer.dequeue(ActiveMQMessageConsumer.java:412)
>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>> org.apache.activemq.ActiveMQMessageConsumer.receive(ActiveMQMessageConsumer.java:531)
>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>> be.meteo.shark.client.emailer.JmsEmailer.processMessages(JmsEmailer.java:178)
>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>> be.meteo.shark.client.emailer.JmsEmailer.main(JmsEmailer.java:409)
>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>> sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>> java.lang.reflect.Method.invoke(Method.java:597)
>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>> org.tanukisoftware.wrapper.WrapperJarApp.run(WrapperJarApp.java:352)
>> INFO | jvm 1 | 2010/08/03 08:35:35 | at
>> java.lang.Thread.run(Thread.java:619)
>>
>> --
>> David Delbecq
>> ICT
>> Institut Royal Météorologique
>> Ext:557
>>
>>
>>
>>
--
David Delbecq
ICT
Institut Royal Météorologique
Ext:557
Re: QueueReceiver.receive(timeout) never return messages
Posted by Dejan Bosanac <de...@nighttale.net>.
Hi David,
did you try using failover transport
http://activemq.apache.org/failover-transport-reference.html
It should take care of detecting network problems and reconnecting.
Then you should just use receive() or message listener
Cheers
--
Dejan Bosanac - http://twitter.com/dejanb
Open Source Integration - http://fusesource.com/
ActiveMQ in Action - http://www.manning.com/snyder/
Blog - http://www.nighttale.net
On Tue, Aug 3, 2010 at 10:56 AM, David Delbecq <da...@oma.be> wrote:
> Hello,
>
> we are having fiability troubles with activeMQ (5.2). To get around those,
> we use in consumer receive(timeout) with a timeout of 60 seconds to detect
> early problems with broker. We assumed if we called receive() on a closed
> connection (broker side closed) we sould somehow get an exception. However,
> when activeMQ server is shutdown for any reason, the receiver never detect
> this and the receive call never return any message. We suspect the timeout
> occurs, but we would expect some kind of exception telling us we need to
> reconnect.
>
>
> Here is consumer code:
> while (!stopNow) {
> Message m = receiver.receive(60000);//wait 60 then next loop
> if (m == null) {
> continue; // timeout?
> }
> if (m instanceof MapMessage) { // The event queue should
> contain only map messages !
> processEventMessage((MapMessage) m);
>
>
> Here is the threaddump of consumer. There are messages in that queue waiting
> since yesterday, but the consumer never received any of those! This is very
> problematic as, for now, we must restart every consumer after restarting the
> broker. We would expect the consumer to be able to detect this and
> reconnect. What's the procedure to follow for this? We could disconnect /
> reconnect every minutes to be sure, but that would mean some kind of
> additional load on the broker.
>
> INFO | jvm 1 | 2010/08/03 08:35:35 | "ActiveMQ Connection Worker:
> tcp://localhost/127.0.0.1:61616" daemon prio=10 tid=0x00007f1cc1d7f000
> nid=0x5d3a waiting on condition [0x0000000041f7b000]
> INFO | jvm 1 | 2010/08/03 08:35:35 | java.lang.Thread.State: WAITING
> (parking)
> INFO | jvm 1 | 2010/08/03 08:35:35 | at sun.misc.Unsafe.park(Native
> Method)
> INFO | jvm 1 | 2010/08/03 08:35:35 | - parking to wait for
> <0x00007f1cc8412618> (a
> java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
> INFO | jvm 1 | 2010/08/03 08:35:35 | at
> java.util.concurrent.locks.LockSupport.park(LockSupport.java:158)
> INFO | jvm 1 | 2010/08/03 08:35:35 | at
> java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:1925)
> INFO | jvm 1 | 2010/08/03 08:35:35 | at
> java.util.concurrent.LinkedBlockingQueue.take(LinkedBlockingQueue.java:358)
> INFO | jvm 1 | 2010/08/03 08:35:35 | at
> java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:947)
> INFO | jvm 1 | 2010/08/03 08:35:35 | at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:907)
> INFO | jvm 1 | 2010/08/03 08:35:35 | at
> java.lang.Thread.run(Thread.java:619)
> INFO | jvm 1 | 2010/08/03 08:35:35 | "WrapperJarAppMain" prio=10
> tid=0x00007f1cc22b8000 nid=0x5f4d in Object.wait() [0x0000000040af6000]
> INFO | jvm 1 | 2010/08/03 08:35:35 | java.lang.Thread.State:
> TIMED_WAITING (on object monitor)
> INFO | jvm 1 | 2010/08/03 08:35:35 | at
> java.lang.Object.wait(Native Method)
> INFO | jvm 1 | 2010/08/03 08:35:35 | - waiting on
> <0x00007f1cc8413280> (a java.lang.Object)
> INFO | jvm 1 | 2010/08/03 08:35:35 | at
> org.apache.activemq.MessageDispatchChannel.dequeue(MessageDispatchChannel.java:77)
> INFO | jvm 1 | 2010/08/03 08:35:35 | - locked <0x00007f1cc8413280>
> (a java.lang.Object)
> INFO | jvm 1 | 2010/08/03 08:35:35 | at
> org.apache.activemq.ActiveMQMessageConsumer.dequeue(ActiveMQMessageConsumer.java:412)
> INFO | jvm 1 | 2010/08/03 08:35:35 | at
> org.apache.activemq.ActiveMQMessageConsumer.receive(ActiveMQMessageConsumer.java:531)
> INFO | jvm 1 | 2010/08/03 08:35:35 | at
> be.meteo.shark.client.emailer.JmsEmailer.processMessages(JmsEmailer.java:178)
> INFO | jvm 1 | 2010/08/03 08:35:35 | at
> be.meteo.shark.client.emailer.JmsEmailer.main(JmsEmailer.java:409)
> INFO | jvm 1 | 2010/08/03 08:35:35 | at
> sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> INFO | jvm 1 | 2010/08/03 08:35:35 | at
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
> INFO | jvm 1 | 2010/08/03 08:35:35 | at
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
> INFO | jvm 1 | 2010/08/03 08:35:35 | at
> java.lang.reflect.Method.invoke(Method.java:597)
> INFO | jvm 1 | 2010/08/03 08:35:35 | at
> org.tanukisoftware.wrapper.WrapperJarApp.run(WrapperJarApp.java:352)
> INFO | jvm 1 | 2010/08/03 08:35:35 | at
> java.lang.Thread.run(Thread.java:619)
>
> --
> David Delbecq
> ICT
> Institut Royal Météorologique
> Ext:557
>
>
>