How to: Implement a BatchMessageListenerContainer for bulk consuming a JMS queue

I recently faced the need for a JMS consumer in Spring Integration - capable of consuming burst of high volume without stressing my target Oracle database with too many commits.

The DefaultMessageListenerContainer does not seem to support anything but message by message transactions.

I googled for solutions and found a couple - but the lot of them suffered from being implemented not by inheritance from DMLC but rather by cloning and modifying the original source code from same - making it vulnerable to break in case I later wish to move to a more recent version of spring-jms. Also the code being cloned referenced private properties of DMLC which consequently had to be left out. And to make it all work also a couple of interfaces and a custom message listener was needed. All in all I did not feel comfortable.

So - what to do?

Best How To :

Well - this is a simple and compact solution that is entirely based on a single class derived from DefaultMessageListenerContainer.

I have only tested with message-driven-channel-adapter and a ChainedTransactionManager though - since this is sort of the basic scenario when needing to do stuff like this.

This is the code:

package dk.itealisten.myservice.spring.components;
import java.util.ArrayList;
import java.util.Enumeration;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
public class BatchMessageListenerContainer extends DefaultMessageListenerContainer {
public static final int DEFAULT_BATCH_SIZE = 100;
public int batchSize = DEFAULT_BATCH_SIZE;
/**
* Override the method receiveMessage to return an instance of BatchMessage - an inner class being declared further down.
*/
@Override
protected Message receiveMessage(MessageConsumer consumer) throws JMSException {
BatchMessage batch = new BatchMessage();
while (!batch.releaseAfterMessage(super.receiveMessage(consumer)));
return batch.messages.size() == 0 ? null : batch;
}
/**
* As BatchMessage implements the javax.jms.Message interface it fits perfectly into the DMLC - only caveat is that SimpleMessageConverter dont know how to convert it to a Spring Integration Message - but that can be helped.
* As BatchMessage will only serve as a container to carry the actual javax.jms.Message's from DMLC to the MessageListener it need not provide meaningful implementations of the methods of the interface as long as they are there.
*/
protected class BatchMessage implements Message {
public ArrayList<Message> messages = new ArrayList<Message>();
/**
* Add message to the collection of messages and return true if the batch meets the criteria for releasing it to the MessageListener.
*/
public boolean releaseAfterMessage(Message message) {
if(message != null) {
messages.add(message);
}
// Are we ready to release?
return message == null || messages.size() >= batchSize;
}
@Override
public String getJMSMessageID() throws JMSException {
return null;
}
@Override
...

Below is a sample showing how it could be used in a Spring application context:

Is this what you want? insert into mytable(name, uuid, . . .) select name, uuid, . . . from mytable where x between $xmin and $xmax and y between $ymin and $ymax; If it is something like this, then you only need one query, just the right conditions....

Well if you disable the scope of the scope that the ng-repeat is on. Then it will no longer render. It essentially becomes static content. This allows you to actually control when it is rendered. ux-datagrid actually uses this concept to turn off dom that is out of view so...

There is no built-in facility to do what you want, at least in PostgreSQL. Doing it effectively would require signifciant changes to how data is stored, as currently each row is independent of all other rows (well, except TOAST pointers for out-of-line stored data that's unchanged in an UPDATE). A...

A "compound index" which is the correct term for your "link" does not create any performance problems on "read" ( since writing new entries is obviously more information ) than an index just on the single field used in the query. With one exception. If you use a "multi-Key" index...

I've written some Rcpp code and R code which works out the binary/Jaccard distance of a binary matrix approx. 80x faster than dist(x, method = "binary"). It converts the input matrix into a raw matrix which is the transpose of the input (so that the bit patterns are in the...

Ubuntu has moved from to Upstart to Systemd in version 15.04 and no longer respects the limits in /etc/security/limits.conf for system services. These limits now apply only to user sessions. The limits for the MySQL service are defined in the Systemd configuration file, which you should copy from its default...

The problem is that you are measuring two different things.164 ms is the time that the database spent executing the query. I suspect the 824 ms that you measured is query execution + instantiation of your Entity objects.

You can import the CSV file into a separate table using mysql LOAD DATA INFILE and then update the entries table using JOIN statement on the basis of similar column name. E.g: update entries a inner join new_table b on a.name = b.name set a.address = b.address ; Here new_table...

Administered objects are also park of web application. You use them to obtain connection (from ConnectionFactory) to JMS provider. You sent message to JMS provider and it routes it to right destination (Destination - topic or queue).

Complete answer: Both. Loading it off of the web will benefit you in a couple of ways: 1) There is a limit to the number of maximum open HTTP requests a browser can have. However, this limit is per domain. So reaching out to google's servers will not prevent you...

I did some benchmarking against three methods. I used an external file for reading (instead of __DATA__). The file consisted of 3 million lines of the exact data you were using. The methods are slurping the file, reading the file line-by line, and using Storable as Sobrique mentioned above. Each...

What is difference between that? AFAIK, both are doing INNER JOIN, First one using a Implicit JOIN syntax whereas the second one using a explicit join syntax I wouldn't expect any performance difference between them but the second style of query using explicit join syntax is much more recommended...

There won't be any difference, since you've only changed the scope of the variables. Since you're not using the variables outside of the scope, the generated bytecode will be identical as well (you can try it out with javap). So use the second style for clarity. Edit: In fact if...

It totally depends on what kind of data you are working with and what kind of searches you want to perform on it. For example, with hash based structures you can not support partial word searches. You could go for an in-memory relational db if your data is really relational...

Hi (i am author of FST): I think the test is flawed: When running your sync (no queues + thread context switches) test in a loop (=proper warmup) I get a Mean of 0.7 micros and Max outlier of 14 micros (doubled number of elements in map though) storing a...

In your case needs value the situation!! When I have this problems the first questions are... The tables that I want to do 'drop table' have few data? If it's yes the best way is the 'drop table' command. If it's no then you need to use the 'insert' and...

The best thing to do would depend on what other fields the table has and what other queries run against that table. Without more details, a non-clustered index on (code, company, createddate) that included the "price" column will certainly improve performance. CREATE NONCLUSTERED INDEX IX_code_company_createddate ON Products(code, company, createddate) INCLUDE...

What happens when you start getting more than single digit IDs? What does 1111 equate to? 1 to 111, 111 to 1 or 11 to 11? Stick with individual fields. It will be easier to interpret, manage and scale. ...

JMS is a API specification that describes how applications can send and receive messages through messaging middle-ware in a standard way and is portable across multiple JMS providers. Without a middle-ware (or broker), just the JMS APIs is of no use. So you need both, broker and JMS API implementation...

I think you're after a particle system. A similar question is here: Drawing many textured particles quickly in OpenGL ES 1.1. Using point sprites is quite cheap, but you have to do extra work in the fragment shader and I'm not sure if GLES2 supports gl_PointSize if you need different...

match and if has different semantics: match is parallel, if is strictly sequential. If you have expression: match expr with | A -> e1 | B -> e2 | C -> e3 | ... Then it may compare branches in any order. In the example, I provided, it may compile...

Whatever comments you supply to this answer, I will continue with helping try revision for you, but was too much to describe in a comment to your original post. You are looking at services within a given UNIX time range, yet your qualifier on service_id in sub-selects is looking against...

Yes, that is okay for a real app. It is typical to emit whole objects from stores even if it just so happens one of the listeners only needs a subset of the data. The idea is to keep it simple and avoid having to change the store when what...

The servlet api provides streams on the request and response. The input stream on the request will only load bytes into the JVM as required. The stream does not hold all the data in memory, a call to read() eventually gets some data out of a small buffer or causes...

Although Rahul is correct, I think it makes sense to add some context to the answer. So, vm stands for virtual machine, which is a very central concept in Java. Java programs execute in the virtual machine. More information about your particular ActiveMQ case, explaining the difference between vm:// and...

The big difference between your code and the example is in the XML config example that myTargetConnectionFactory is actually a bean managed by Spring. You aren't doing that. You are just creating a new object Spring doesn't know about. The magic happens when setting the targetConnectionFactory of myConnectionFactory. Even though...

Except for trivial queries, there is no way to know if you have the optimal query & indexes. Nor can you get a metric for how well designed the schema and application are. 3 seconds on a cold system for a 3-way JOIN with "Rows" of 409, 45, 1 is...

Click this jboss link; which contains quickstart program which resolves mentioned problem. 1) Download the code; Import pom.xml of run jboss-helloworld-jms in netbeans. Run clean build. 2) Use this link to resolve maven build problems: Pom not found 3) Following is snippet which used to connect to server private static...

My guess is that you also have an index on MachineName - or that SQL is deciding that since it needs to group by MachineName, that would be a better way to access the records. Updating statistics as suggested by AngularRat is a good start - but SQL often maintains...

You may want to refer to specific JNDI Reference for specific versions. JBOSS AS 7.2 is covered here: https://docs.jboss.org/author/display/AS72/JNDI+Reference (note that in JBOSS AS 7.x, jnp is no longer supported, older JBOSS versions do support the jnp:// and access via the standard naming services). Another link: https://access.redhat.com/documentation/en-US/JBoss_Enterprise_Web_Platform/5/html/Administration_And_Configuration_Guide/Naming_on_JBoss-The_Naming_InitialContext_Factories.html. ...

Basically combine() takes an array with the values to combine and a size of the wanted combination results sets. The inner function c() takes an array of previously made combinations and a start value as index of the original array for combination. The return is an array with all made...

With respect to the modeling question, I'd like to offer a fourth alternative, which is, in fact, a mix of your options 1 and 2: introduce a separate class (hierarchy) for these 'excluded/missing' symptoms, diseases or treatments, and have the specific exclusions as instances: :Exclusion a owl:Class . :ExcludedSymptom rdfs:subClassOf...

Files.list() is a O(N) operation whereas sorting is O(N log N). It is far more likely that the operations inside the sorting which matter. Given the comparisons don't do the same thing, this is the most likely explanation. There is a lot of files with the same modification date under...

microtime(true) is your PHP friend. If there is an index starting with gameid, "Approach 1" might be faster, since it can do the COUNT(*) in the index. On the other hand, there are two things that count against it: 3 queries is usually slower than 2. If there are a...

You need a composite index for posts_to_tribes: INDEX(tribe_id, post_id). The GROUP BY was to compensate for the JOIN exploding the number of rows. Here is a better workaround, than IN ( SELECT ... ): SELECT p.post_id, p.date_created, p.description, p.last_edited, p.link, p.link_description, p.link_image_url, p.link_title, p.total_comments, p.total_votes, p.type_id, p.user_id FROM posts p...

You can get a total average number of events per second like this: #!/usr/bin/env python3 import time import datetime as dt start_time = dt.datetime.today().timestamp() i = 0 while(True): time.sleep(0.1) time_diff = dt.datetime.today().timestamp() - start_time i += 1 print(i / time_diff) Which in this example would print approximately 10. Please note...

Your foreach is like a Select. var ProductCodes = db.tbl_ProdCodeValues.Where(x=>x.productID == _Product.productID); var withMatches = ProductCodes .Select(code => new { code, matches = db.InventoryCodeValues.Any(x => x.InventoryValue.ToLower().Contains(code.ProdCodeValue)) }); And now all of this remotes to the database. Look at the query plan to see whether this is acceptable already or whether...

Pygame doesn't have the ability to do this natively. If you really want this, you'll need to brush up on your trigonometry to map lines from the 3D space to the 2D screen. At that point, you'll essentially be re-implementing a 3D engine.

You are comparing a varchar column with a numeric literal (245643). This forces Oracle to convert one side of equality, and off hand, it seems as though it's choosing the "wrong" side. Instead of having to guess how Oracle will handle this conversion, use a character literal: SELECT * FROM...