You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@camel.apache.org by Dimitry Pletnikov <di...@pletnikov.com> on 2013/12/11 05:58:26 UTC
Understanding ActiveMQ consumer with concurrentConsumers
Using:
- ServiceMix 4.3.5
- ActiveMQ 5.7.0
- Camel 2.10.7
I have a persistent queue in ActiveMQ, which I consume messages from
using activemq-component with concurrentConsumers=20 set.
ActiveMQ connections are pooled:
<bean id="activemqConnectionFactory"
class="org.apache.activemq.ActiveMQConnectionFactory"
depends-on="broker">
<property name="brokerURL"
value="vm://${activemq.brokername}?create=false&waitForStart=10000"
/>
<property name="userName" value="${activemq.user}"/>
<property name="password" value="${activemq.password}"/>
</bean>
<bean id="pooledConnectionFactory"
class="org.apache.activemq.pool.PooledConnectionFactory">
<property name="idleTimeout" value="0" />
<property name="maxConnections" value="8" />
<property name="connectionFactory" ref="activemqConnectionFactory" />
</bean>
<bean id="resourceManager"
class="org.apache.activemq.pool.ActiveMQResourceManager"
init-method="recoverResource">
<property name="transactionManager" ref="recoverableTxManager" />
<property name="connectionFactory" ref="activemqConnectionFactory" />
<property name="resourceName" value="activemq.${name}" />
</bean>
<reference id="recoverableTxManager"
interface="org.apache.geronimo.transaction.manager.RecoverableTransactionManager"
availability="mandatory" />
I have a route looking like:
<from uri="activemq://item_updates?concurrentConsumers=20" />
<delay><constant>5000</constant></delay>
<log message="item ${header.item_id}" />
I expect to see something like:
item 1
item 2
item 3
item 4
item 5
<5 seconds pause>
item 6
item 7
..etc
Instead the output looks like:
item 1
<5 seconds pause>
item 2
<5 seconds pause>
..etc
My guesses are:
- all messages are being prefetched by the first concurrent consumer?
-or-
- it has something to do with pooled ActiveMQ connections?
-or-
- ???
Looking at examples and documentation it should be simple and easy. In
my case it's not.
Re: Understanding ActiveMQ consumer with concurrentConsumers
Posted by dimitry12 <di...@pletnikov.com>.
I tested this morning and can confirm: it was a prefetch-related issue.
I modified the route as:
<from
uri="activemq://item_updates?concurrentConsumers=20&destination.consumer.prefetchSize=1"
/>
<delay><constant>5000</constant></delay>
<log message="item ${header.item_id}" />
and now each of 20 concurrent consumers gets a message first then they all
"delay" together.
The relevant piece, which I was missing, is mentioned on
http://activemq.apache.org/what-is-the-prefetch-limit-for.html - the default
prefetch limit for persistent queues (my case) is 1,000. I had less than
1,000 messages in the queue, so they were all prefetched by the first of 20
concurrent consumers.
--
View this message in context: http://camel.465427.n5.nabble.com/Understanding-ActiveMQ-consumer-with-concurrentConsumers-tp5744590p5744621.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Understanding ActiveMQ consumer with concurrentConsumers
Posted by dimitry12 <di...@pletnikov.com>.
@kraythe: Yes, you described what I was expecting: every consumer getting its
own message and processing it simultaneously. What I was getting: one
consumer prefetching all messages in the queue and processing them
sequentially, while all other consumers were sitting and doing nothing.
--
View this message in context: http://camel.465427.n5.nabble.com/Understanding-ActiveMQ-consumer-with-concurrentConsumers-tp5744590p5744623.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Understanding ActiveMQ consumer with concurrentConsumers
Posted by "kraythe ." <kr...@gmail.com>.
Each of the consumer routes is brought up concurrently on different
threads. They will suck in messages as fast as they can, competing like
kids for candy tossed on the middle of a floor. So while thread 1 is
waiting 5 seconds, threads 2, 3, 4, 5, ... and so on aren't and grabbing
the next message on the queue. When one of those is waiting, the others are
probably not. The more concurrent consumers you put on the queue, the less
effect your delay timer will have.
Out of curiosity why are you doing this? There are better ways of
throttling if that is your goal.
*Robert Simmons Jr. MSc. - Lead Java Architect @ EA*
*Author of: Hardcore Java (2003) and Maintainable Java (2012)*
*LinkedIn: **http://www.linkedin.com/pub/robert-simmons/40/852/a39
<http://www.linkedin.com/pub/robert-simmons/40/852/a39>*
On Tue, Dec 10, 2013 at 10:58 PM, Dimitry Pletnikov
<di...@pletnikov.com>wrote:
> Using:
> - ServiceMix 4.3.5
> - ActiveMQ 5.7.0
> - Camel 2.10.7
>
> I have a persistent queue in ActiveMQ, which I consume messages from
> using activemq-component with concurrentConsumers=20 set.
>
> ActiveMQ connections are pooled:
>
> <bean id="activemqConnectionFactory"
> class="org.apache.activemq.ActiveMQConnectionFactory"
> depends-on="broker">
>
> <property name="brokerURL"
> value="vm://${activemq.brokername}?create=false&waitForStart=10000"
> />
> <property name="userName" value="${activemq.user}"/>
> <property name="password" value="${activemq.password}"/>
> </bean>
>
> <bean id="pooledConnectionFactory"
> class="org.apache.activemq.pool.PooledConnectionFactory">
> <property name="idleTimeout" value="0" />
> <property name="maxConnections" value="8" />
> <property name="connectionFactory" ref="activemqConnectionFactory"
> />
> </bean>
>
> <bean id="resourceManager"
> class="org.apache.activemq.pool.ActiveMQResourceManager"
> init-method="recoverResource">
> <property name="transactionManager" ref="recoverableTxManager" />
> <property name="connectionFactory" ref="activemqConnectionFactory"
> />
> <property name="resourceName" value="activemq.${name}" />
> </bean>
>
> <reference id="recoverableTxManager"
>
> interface="org.apache.geronimo.transaction.manager.RecoverableTransactionManager"
> availability="mandatory" />
>
> I have a route looking like:
>
> <from uri="activemq://item_updates?concurrentConsumers=20" />
> <delay><constant>5000</constant></delay>
> <log message="item ${header.item_id}" />
>
> I expect to see something like:
> item 1
> item 2
> item 3
> item 4
> item 5
> <5 seconds pause>
> item 6
> item 7
> ..etc
>
> Instead the output looks like:
> item 1
> <5 seconds pause>
> item 2
> <5 seconds pause>
> ..etc
>
> My guesses are:
> - all messages are being prefetched by the first concurrent consumer?
> -or-
> - it has something to do with pooled ActiveMQ connections?
> -or-
> - ???
>
> Looking at examples and documentation it should be simple and easy. In
> my case it's not.
>