Is CouchbaseQueue API in http://docs.couchbase.com/sdk-api/couchbase-java-client-2.5.5/com/couchbase/client/java/datastructures/collections/CouchbaseQueue.html thread safe? Or can i achieve the similar behavior by Bucket.getAsync().queuePop()?
I want to access this async behavior from multiple JVM and JVM’s don’t know about each other. I do not want to go with Kafka path at this moment. Can anyone give some light as what will be the best way to achieve async queue behavior or refer me to appropriate section?
Yes, CouchbaseQueue is safe for concurrent use by multiple threads and multiple JVMs.
The Java class itself has no mutable state. The backing array document is modified using subdocument operations (which are atomic) and optimistic locking. It should perform reasonably well as long as there’s not a huge amount of contention. Here’s the source code if you’d like to see for yourself.
A couple of things to keep in mind regarding the implementation as of SDK 2.6.1:
- The API for this class is not committed, meaning it could change at any time.
- CouchbaseQueue.poll() throws
ConcurrentModificationExceptionif there’s too much contention and the “compare and swap” retry limit is exceeded. The retry limit is controlled by a Java System Property
com.couchbase.datastructureCASRetryLimit, and defaults to
EDIT: I see you were asking specifically about SDK 2.5.5 – CouchbaseQueue is the same in 2.5.5 and 2.6.1.
The behavior is the same, but the implementation is a bit different.
CouchbaseQueue.poll() uses a subdocument operation to get just the head of the queue, while
AsyncBucket.queuePop() loads the entire document. I’m not sure if this is intentional, or an opportunity for applying the same optimization in
queuePop(). What do you think, @subhashni?
Both collections and bucket API use full document get and then a subdocument remove as subdocument remove operation does not return the value removed, the document cas is used to make sure there wasn’t any concurrent modification between the two operations.
Yes, it is documentation bug for the collections one, the bucket API is committed.
Thanks for the reply @david.nault, @subhashni and @ingenthr. I will try a POC with high load against the “optimistic locking”. Want to build a Producers/Consumers paradigm with flexibility of manipulating Queue. Exploring CouchbaseQueue/CouchbaseMap if it could be good enough to serve the purpose. Will update my findings as it could help others too.