Using a LinkedBlockingQueue and flush to mysql

Question

Would a linkedblockingqueue be suitable for the following:

1. insert strings (maximum 1024 bytes) into the queue at a very high rate
2. every x inserts or based on a timed interval, flush items into mysql

During the flush, I was looking at the API: http://docs.oracle.com/javase/1.5.0/docs/api/java/util/concurrent/LinkedBlockingQueue.html

At was wondering it drainTo would be a good choice, since I have to aggregate before flushing.

So I would drainTo the items in the queue, then iterate and aggreate and then write to mysql.

Will this be suitable for upto 10K writers per second?

Do I need to consider any locking/synchronization issues or is that taken care of already?

I will store this linkedblockingqueue as the value in a concurrenthashmap.

Items will never be removed from the hashmap, only inserted if not present, and if present, I will append to the queue.

Answer 1

It depends a bit if the inserter is per queue or for all queues. If I am understanding your spec, I would think something like the following would work.

Writer adds an item to the one of the LinkedBlockingQueue collections in your map. If the size of the queue is more than X (if you want it per queue) then it signals the MySQL inserter thread. Something like this should work:

queue.add(newItem);
// race conditions here that may cause multiple signals but that's ok
if (queue.size() > 1000) {
    // this will work if there is 1 inserter per queue
    synchronized (queue) {
        queue.notify();
    }
}
...

Then the inserter is waiting on the queue and in something like the following loop:

List insertList = new ArrayList();
while (!done) {
    synchronized (queue) {
        // typically this would be while but if we are notified or timeout we insert
        if (queue.size() < 1000) {
            queue.wait(MILLIS_TIME_INTERVAL);
        }
    }
    queue.drainTo(insertList);
    // insert them into the db
    insertList.clear();
}

It gets a bit more complicated if there 1 one thread doing the inserts across all queues. I guess the question is then why do you have the ConcurrentHashMap at all? If you do have 1 inserter which, for example, is inserting into multiple tables or something then you will need a mechanism to inform the insert which queue(s) need to be drained. It could just run through all of the queues in the map but that might be expensive. You would synchronize on some global lock object or maybe the map object instead of the queue.

Oh, and as @Peter Lawrey mentioned, you will quickly run out of memory if your database is slower than the writers so make sure the queues have a proper capacity set so they limit the writers and keep the working memory down.

Hope this helps.