Achieving fast consumption rate

[stephanion2002]

I am not using spring integration yet.

Here is the problem: I have 20 exchanges, and I want to make 4 different type of queues(email/http/sms/ftp) and 4 different types of handlers(email/http/sms/ftp) for each one of those 20 exchanges....

I am getting a topicName from outside the system, so I create one exchange , i.e. (topic+"FanOutExchange") like this:

FanoutExchange exchange = new FanoutExchange(topicDbObj.getName()+Constants.EXCH ANGE);

Then since I have 4 different protocols of subscriber's subscriptions(email/http/sms/ftp), I create 4 queues like this:
Queue queue = new Queue(subObj.getProtocol()+Constants.QUEUE);

And then I bind them together like this:

Binding binding = new Binding(queue, exchange);

This adds to 20 exchanges and corresponding 80 queues.

Let's move onto POJO handlers' side...

I want to have 4 POJO handlers, just Like I have 4 queues, for each exchange, but the irony is this that these 4 POJO handlers should be nothing but different instances of the same 4 classes..How Can I achieve such a setting?

Can Spring integration help me here?

[stephanion2002]

The listener container has a 'concurrentConsumers' property. Here's the JavaDoc from that setter:

/**
 * Specify the number of concurrent consumers to create. Default is 1.
 * <p>
 * Raising the number of concurrent consumers is recommendable in order to
 * scale the consumption of messages coming in from a queue. However, note
 * that any ordering guarantees are lost once multiple consumers are
 * registered. In general, stick with 1 consumer for low-volume queues.
 */

Hi Mark,

My Handler POJO does not have any specific instance variables initializations except some DIs. If I set concurrent consumers property=20 , then does that mean I will have 20 "different" objects which will not intermingle with the same data ? Do I have to do something else to make this part of my application "thread-safe" ? Should I make my handler POJO's scope=protoype for this ?

Thanks,
Kshitiz

[Mark Fisher]

The concurrentConsumers value will not cause multiple instances of your listener's delegate object to be created. It does mean that multiple threads may be invoking that object concurrently.

[stephanion2002]

Thanks Mark

[stephanion2002]

Hi,
Few days back, I had posted this query:

Thanks Mark,

I am using a message Handler which is just a POJO, thanks to Spring AMQP , but now I am facing a situation. I want to hold acknowledging the message so that it stays live on the rabbitMQ till my POJO's handle message gets completed successfully.

On rabbitMQ mailing list , I have been told to consume with no_ack=false and then send a basic.ack after the processing is complete. How can we achieve that from this POJO or its configuration ?

Regards,
Kshitiz Garg

Dave Syer from springsource had replied to the above quote:

We have transactions covered fully in the RabbitTemplate now, but the asynch use case (message listeners) is not finished yet. However, I think if you set autoAck=false in the message listener container you should see the acks going after the POJO listener has completed

In my case, I am starting another thread which completes an email sending work, and I want to explicitly set basic.ack=true from this new thread so that I can be sure that messages are going to be deleted from RabbitMQ only after the completion of the message processing.

Is there any workaround for this ? Desperately waiting for an answer

Many Thanks,
Kshitiz Garg

[stephanion2002]

Hi ,

Few days back, I had posted the query in the previous post.

Kindly help.

Many Thanks,
Kshitiz Garg

[Dave Syer]

The basic async use cases are implemented on master now. But sending an ACK from a different thread is not covered - it's not a mainstream Spring TX use case, and it's bound to have consequences (e.g. we can't guarantee that by the time you send the ACK the channel was not already closed). If you open a JIRA we can discuss the requirement in more detail there.

[stephanion2002]

Thanks Dave,

I will be opening a JIRA (in the scope of Spring AMQP)

Regards,
Kshitiz Garg