How to set ActiveMQ capacity

[mberg]

Hello all,

I have a scenario where a client application sends messages to activeMQ using Spring Integration. I have some processors that consume these messages, but they are not quite fast enough to handle the load. So I want to throttle the client a bit, by assigning a max capacity to the activeMQ queue. When the capacity is reached, the client should simply wait until capacity is available again.

I'm not sure how to achieve this. Let's start with the spring configuration:

	<si:channel id="fromClient" />

	<si:gateway id="echoGateway"
		service-interface="com.mypackage.EchoGateway"
		default-request-channel="fromClient"
	/>

	<int-jms:outbound-channel-adapter channel="fromClient" destination="fromClientQueue" />

	<bean id="fromClientQueue" class="org.apache.activemq.command.ActiveMQQueue">
		<constructor-arg value="queue.fromClient"/>
	</bean>

Right, so I tried this:

	<si:channel id="fromClient" >
	    <si:queue capacity="100"/>
	</si:channel>

However, the capacity is ignored and the activeMQ queue is flooded with messages as soon as the client starts.

I looked at the ActiveMQQueue class, but found no relevant getters or setters. I also tried in the web interface of activeMQ, but you can only observe the queues there.

What do I do to solve this problem?

[Gary Russell]

Not sure what you mean by "capacity is ignored" but you didn't show the full config after you added the queue element. A queue channel needs a poller.

Googling "activemq throttle client" returned this link...

http://activemq.apache.org/producer-flow-control.html

[mberg]

Hello Gary,

When I said that "capacity is ignored", I meant that my gateway method is able to post millions of messages to the channel. It doesn't stop when there are 100 messages on the queue.

As for the poller, I have this global definition:

	<si:poller id="poller" default="true" fixed-delay="1000"/>

Thanks for the link. However, the features described relate to throttling of clients based on memory and disk constraints. While something like this could work for me, I was hoping for a way to throttle based on the actual queue size. I simply want to place a constraint on the queue capacity, so that whenever X messages are waiting on the queue, then when the gateway method is called, it will actually wait for capacity to become available before returning. Is this possible?

[Gary Russell]

First of all, "fromClient" is not a "queue", it's a channel (although in this case, a QueueChannel). It has no bearing on the real JMS queue "queue.fromClient".

JMS doesn't have any standard notion of flow control; that's up to the broker (ActiveMQ etc). Setting a limit on the channel's queue has no bearing on whether or not ActiveMQ will stop a message from being posted. Spring Integration (via the poller in this case) will pump out as many messages as the broker will accept - it's up to the broker to impose any flow control - Spring Integration has no control over it. You have to explore your broker of choice for how to deal with this situation.

Hope that helps.

[mberg]

Hello Gary,

Apologies for not using the correct terminology, I realize the difference between a queue and a channel.

Regarding the limit on the queue, I can see now how this limit is irrelevant to JMS. JMS is taking messsages off the queue and sending them to the activeMQ broker at a rate where the SI queue limit is never reached.

I'll try to look into which configuration options are available in ActiveMQ. It must be possible somehow to configure that a queue should have a cap on the number of queued messages.

Thanks again for your help!

[Gary Russell]

JMS is taking messsages off the queue and sending them to the activeMQ broker at a rate where the SI queue limit is never reached.

Just to clarify, the poller is taking messages off the QueueChannel and sending them to JMS using the outbound adapter.