title: Mitigating DB WRITE contention from concurrent messaging system
body: <p><strong>Please note:</strong> although my question here specifically involves Camel (2.11.0) and ActiveMQ (5.8.0), its really about the proper design of a concurrent messaging solution, and could very conceivably be answered by anybody with any formidable messaging and/or concurrency experience.</p>
  
  <p>I have a Camel route that starts by reading messages off of an ActiveMQ queue (<code>myQueue</code>) and sending them to a bean (<code>processorBean</code>) for processing:</p>
  
  <pre><code>&lt;camelContext id="my-camel-context" xmlns="http://camel.apache.org/schema/spring"&gt;
      &lt;endpoint id="myQueue" uri="myBroker01:queue:myQueue" /&gt;
  
      &lt;route id="my-route"&gt;
          &lt;from ref="myQueue" /&gt;
          &lt;to uri="bean:processorBean?method=process" /&gt; 
      &lt;/route&gt;
  &lt;/camelContext&gt;
  </code></pre>
  
  <p>And:</p>
  
  <pre><code>public class ProcessorBean {
      public void process(Exchange exchange) {
          String messageJSON = (String)exchange.getIn().getBody();
  
          // Example: now messageID might be "12345"
          String messageID = parseJSON(messageJSON, "messageID");
  
          // Look up DB records based on this messageID.
          // The same messageID will *always* return the same list of widgets.
          List&lt;Widget&gt; widgets = dao.getWidgetsByMessageID(messageID);
  
          // Make updates to widgets.
          for(Widget widget : widgets) {
              widget.setFizz(true);
              widget.setBuzz("Yahtzee!!!");
          }
  
          // Persist all updates to the widget list.
          dao.updateAll(widgets);
      }
  }
  </code></pre>
  
  <p>This bean uses the message's ID (a String <code>messageID</code> field) to look up a bunch of records in a DB, make a change to them, and then saves them. It's also <strong>very</strong> important to note that the messages arrive at <code>myQueue</code> from an external process that I have no control over. In other words, I can't prevent messages with identical <code>messageID</code> values (hereafter "duplicates" or <em>duplicate messages</em>) from showing up on the thread. Hence this external process could send 1,000 messages onto <code>myQueue</code>, and 20 of them could all have <code>messageID=12345</code>.</p>
  
  <p>Currently, I only have 1 Camel consumer configured to be running (hence it's "single threaded"). So there currently isn't any harm (besides possible unnecessary performance issues) when duplicates show up. Each message gets processed, one at a time, and if there are 20 messages with the same <code>messageID</code>, well then, the same DB records get the same exact (unnecessary) updates over and over again. Bad for performance, sure, but it doesn't create "bad data", dirty writes, our produce race conditions, etc.</p>
  
  <p>I now want to add more Camel consumer threads to the equation, so that there might be, say, 10 consumer threads all reading off of <code>myQueue</code>.</p>
  
  <p>Obviously, now we have a potential for WRITE contention in the DB. Say there are 2 messages on <code>myQueue</code>, and both have <code>messageID=12345</code>. One Camel consumer thread reads the 1st message, and another thread reads the 2nd message at the same time, or thereabouts. Each thread routes its message to its own copy/version of <code>processorBean</code>. Both <code>processorBean</code> instances execute around the same time, use the <code>messageID</code> to READ the same records out of the DB, performs the same operations to them in-memory, and then calls <code>dao.updateAll(...)</code> to WRITE the changes to the same records, at the same time. If both threads are updating the same DB records at the same time, there will be contention.</p>
  
  <p>Another important note is that changing the DB (controlled by another team) to implement things like sharding, optimistic locking, etc., is not an option in this case (too long of a backstory).</p>
  
  <p>My question: <strong>what can be done at the Java layer to mitigate WRITE contention in this situation?</strong> The WRITE contention <em>must</em> be dealt with from inside the application. Thoughts?</p>