Problem
The objective is to create a lightweight object pool that should be as simple as possible, but not simpler. It should be able to create and pool any type of object, be configurable (i.e. pool size) and be able to be replenish exhausted pools or timeout waiting for returned objects. It should be thread safe and able to operate at high volumes (borrow, use and return 300 objects per second). I should mention that the overall objective is to make the use of expensive to create objects more performant.
Based on a number of articles and personal experience I have written the following code, and the code has its own Github repository. I welcome any comment regarding this code but in particular I would appreciate feedback on the following points:
- Is there a glaring issue with the code not satisfying the above specified concerns?
- Concurrency not being my strong point, I suspect that there are issue related to threading. Please let me know if you find any.
- Comments regarding style, naming, structure etc are very welcome.
- What features are missing that you would like to see in a simple object pool?
An object pool must implement the AbstractObjectPool class:
public abstract class AbstractObjectPool<T> implements ObjectPool {
private AbstractQueue<T> pool;
/**
* Initialise the pool and populate it with poolSize number of objects
*
* @param poolSize the size of object to initialise the pool
* @param pool the abstract queue pool
*/
protected void initialize(int poolSize, AbstractQueue<T> pool) {
setPool(pool);
for (int i = 0; i < poolSize; i++) {
pool.add(createObject());
}
}
/**
* Gets the pool
*
* @return AbstractQueue pool object
*/
public AbstractQueue<T> getPool() {
return pool;
}
/**
* Sets the pool.
*
* @param pool the pool to set
*/
private void setPool(AbstractQueue<T> pool) {
this.pool = pool;
}
/**
* Gets the next free object from the pool.
* <p/>
* Different strategies can be implemented to deal with a
* situation where the pool doesn't contain any objects.
* <p/>
* Some possible options:
* <p/>
* 1. a new object will be created and given to the caller of this method.
* 2. a PoolDepletionException is thrown
* 3. wait for a specified time for an object to be returned
*
* @return T borrowed object
* @throws PoolDepletionException thrown if the pool has been depleted
* @throws InterruptedException
*/
public abstract T borrowObject() throws PoolDepletionException, InterruptedException;
/**
* Returns object back to the pool.
* <p/>
* Possible implementation may include code to clean/reset the
* object to initial values.
*
* @param object object to be returned
*/
public void returnObject(T object) {
if (object == null) {
return;
}
this.pool.offer(object);
}
/**
* Creates a new object.
*
* @return T new object
*/
protected abstract T createObject();
/**
* Destroys an object.
*
* @param pool the pool to destroy
*/
protected void destroyObject(AbstractQueue<T> pool) {
T object = pool.poll();
object = null;
}
/**
* Destroys the entire pool.
*/
public void destroyPool() {
while (pool != null && !pool.isEmpty()) {
destroyObject(pool);
}
pool = null;
}
}
Here is an implementation for the BOON JSON parser object (https://github.com/boonproject/boon/wiki/Boon-JSON-in-five-minutes).
public class JsonParserFixedPool extends AbstractObjectPool<JsonParserAndMapper> {
private LinkedBlockingQueue<JsonParserAndMapper> pool = new LinkedBlockingQueue<>();
private int pollTimeout = 3000;
public JsonParserFixedPool() {
initialize(20, pool);
}
public JsonParserFixedPool(int poolSize) {
initialize(poolSize, pool);
}
public JsonParserFixedPool(int poolSize, int pollTimeout) {
initialize(poolSize, pool);
this.pollTimeout = pollTimeout;
}
/**
* Borrow one BOON JSON parser object from the pool.
*
* @return returns one BOON JSON parser object
* @throws PoolDepletionException thrown if there are no BOON JSON parser objects in the pool
* @throws InterruptedException
*/
@Override
public JsonParserAndMapper borrowObject() throws PoolDepletionException, InterruptedException {
JsonParserAndMapper jsonParserAndMapper = pool.poll(this.pollTimeout, TimeUnit.MILLISECONDS);
if (jsonParserAndMapper == null) {
throw new PoolDepletionException("The JSON parser pool is empty and was not replenished within timeout limits.");
}
return jsonParserAndMapper;
}
/**
* Creates a BOON JSON parser object.
*
* @return a new BOON JSON parser object
*/
@Override
protected JsonParserAndMapper createObject() {
return new JsonParserFactory().useAnnotations().usePropertiesFirst().create();
}
}
And this is how you would use it:
// Create pool of 20 JSON Parser objects
JsonParserFixedPool jsonParserFixedPool = new JsonParserFixedPool(20);
// Borrow a JSON object from the pool
JsonParserAndMapper jsonParserAndMapper = jsonParserFixedPool.borrowObject();
// Return the object to the pool after using it
jsonParserFixedPool.returnObject(jsonParserAndMapper);
In this implementation the request to borrow an object will poll for 3000 milliseconds before throwing a DeletionException.
An alternative implementation would be to execute a monitor in a separate thread that attempts to maintain the pool size by creating/destroying objects in the pool.
Here is the code for the monitor:
public abstract class FlexibleObjectPool<T> extends AbstractObjectPool<T> {
private ScheduledExecutorService executorService;
protected FlexibleObjectPool() {
provokePoolMonitor(10, 20, 3000);
}
protected FlexibleObjectPool(int minIdle, int maxIdle, int validationInterval) {
provokePoolMonitor(minIdle, maxIdle, validationInterval);
}
/**
* Checks pool conditions in a separate thread.
* <p/>
* If pool size rises to above maxIdle then the number of objects
* deleted is equal to the excess over the minIdle.
* <p/>
* If pool size drops below minIdle new objects are created to
* bring it up to the minIdle level.
*
* @param minIdle minimum number of objects idling in the object pool
* @param maxIdle maximum number of objects idling in the object pool
* @param validationInterval time in milli-seconds for periodical checking of minIdle / maxIdle conditions in a separate thread.
* When the number of objects is less than minIdle, missing instances will be created.
* When the number of objects is greater than maxIdle, excess instances will be removed.
*/
protected void provokePoolMonitor(final int minIdle, final int maxIdle, int validationInterval) {
executorService = Executors.newSingleThreadScheduledExecutor();
executorService.scheduleWithFixedDelay(() -> {
int size = getPool().size();
if (size < minIdle) {
int sizeToBeAdded = minIdle - size;
for (int i = 0; i < sizeToBeAdded; i++) {
getPool().add(createObject());
}
} else if (size > maxIdle) {
int sizeToBeRemoved = size - maxIdle;
for (int i = 0; i < sizeToBeRemoved; i++) {
destroyObject(getPool());
}
}
}, validationInterval, validationInterval, TimeUnit.MILLISECONDS);
}
/**
* Shutdown this pool's executor service and deletes the object pool.
*/
public void destroyPool() {
if (executorService != null) {
executorService.shutdown();
executorService = null;
}
// Destroys the entire pool.
super.destroyPool();
}
}
And the implementation for a JSON pool:
public class JsonParserFlexiblePool extends FlexibleObjectPool<JsonParserAndMapper> {
private LinkedBlockingQueue<JsonParserAndMapper> pool = new LinkedBlockingQueue<>();
private int pollTimeout = 3000;
public JsonParserFlexiblePool() {
initialize(20, pool);
}
public JsonParserFlexiblePool(int poolSize) {
initialize(poolSize, pool);
}
public JsonParserFlexiblePool(int poolSize, int pollTimeout) {
initialize(poolSize, pool);
this.pollTimeout = pollTimeout;
}
public JsonParserFlexiblePool(int poolSize, int minIdle, int maxIdle, int validationInterval) {
super( minIdle, maxIdle, validationInterval);
initialize(poolSize, pool);
}
/**
* Borrow one BOON JSON parser object from the pool.
*
* @return returns one BOON JSON parser object
* @throws PoolDepletionException thrown if there are no BOON JSON parser objects in the pool
* @throws InterruptedException
*/
@Override
public JsonParserAndMapper borrowObject() throws PoolDepletionException, InterruptedException {
JsonParserAndMapper jsonParserAndMapper = pool.poll(this.pollTimeout, TimeUnit.MILLISECONDS);
if (jsonParserAndMapper == null) {
throw new PoolDepletionException("The JSON parser pool is empty and was not replenished within timeout limits.");
}
return jsonParserAndMapper;
}
/**
* Creates a BOON JSON parser object.
*
* @return a new BOON JSON parser object
*/
@Override
protected JsonParserAndMapper createObject() {
return new JsonParserFactory().useAnnotations().usePropertiesFirst().create();
}
}
And we use it like so:
// Create pool of 10 JSON Parser objects, with a min of 10 and max of 15 that executes the monitor every 2 seconds
jsonParserFlexiblePool = new JsonParserFlexiblePool(10, 10, 15, 2000);
// Borrow a JSON object from the pool
JsonParserAndMapper jsonParserAndMapper = jsonParserFixedPool.borrowObject();
// Return the object to the pool after using it
jsonParserFixedPool.returnObject(jsonParserAndMapper);
Solution
I have had a long look through your code. I like that you have tried to make a generic, type-safe pool. It is a problem that I have solved a number of times in the past (sometimes well, sometimes not so well), and I like looking at other people’s solution to the problem.
The basic concept in your code looks like it will work, but it is just “by the skin of your teeth” working.
First up, I am going to rattle off a long list of things that concern me, then I am going to suggest a few things that will significantly change the structure of the code, and, perhaps at the end I will put out an alternative solution that will maybe give you some ideas you can (re)use.
AbstractObjectPool -> Head-Scratchers
-
What’s
ObjectPool– the interface you implement, but don’t put in your question. On GitHub it is a@FunctionalInterfacewith a singleT borrowObject()method. You should have included this in your question! -
Why do you need a functional interface at all? What problem does it solve? It is not like you can use the
AbstractObjectPoolor any derivative in a way where the interface can be taken advantage of. -
On the other hand, why don’t you have an interface that defines the pool concept completely, with both a borrow, and a return method?
-
In your Abstract class, why do you have the initialization set up as an initialize method, and not a constructor?
-
Why do you have your
poolinstance set up as anAbstractQueueand not just the interfaceQueue? When possible you should declare variables by the interface… Ohh, you’ve usedAbstractQueueto get thepoll()andoffer()methods, but to do that you should use theBlockingQueueinterface from the concurrent package. -
Why do you have a
getPool()method (and why does it return the internalpool)? That queue should be “safe” from external meddlers, and you expose it this way so that someone can do:pool.getPool().clear()and suddenly all your cached items are gone! -
Why is the
borrowObject()method abstract? Why can’t you implement it in this class, and just leave thecreateObjectas an abstract method? -
On the topic of
borrowObject, why is it calledborrowObjectand not justborrow? If the pool is a pool of JSON parsers, why would you borrow an Object? The semantics ofparserpool.borrow()is much better thanparserpool.borrowObject(). The same is true forreturn(). -
Why do you need ‘destroy*’ methods? What are they for? Their implementation is partially broken anyway because they set the
poolto null and then subsequent usage will throw ugly NullPointerExceptions. If you expect your pools to have an orderly shut-down process, then you should track the shutdown state, and throw appropriate exceptions for calls to an already shut down pool (thinkClosable) -
Your
destroyObject(...)method is all broken, it takes in apoolas an argument, but thatpoolmay be different to the one the class manages. Why does it need that input argument? Regardless, all it is doing is “popping” an item off the stack.
Having listed the above concerns, there are some things missing which are also concerning. For a start, there’s nothing in the AbstractQueue that manages concurrency (but there is in the BlockingQueue. It is your stated goal to have a thread-safe operation, but you have nothing so far that actually supports that – you completely rely on the implementing classes to implement the right systems to support thread-safe operation. An implementing class could initialize your class with a simple PriorityQueue, and then you’re borked.
Implementations -> Head-Scratchers
In your various implementing classes, you have a number of other head-scratchers:
- You have a shadow-version of the
poolvariable in both the abstract, and the implementing classes. Which pool do you think is used in the different contexts? - the destroyPool code is still messy, and, I think, unnecessary.
Suggestions
Right, the first major suggestion is to consider the use of protected constructors for initialization, not using an initializing method.
Next up, you should be using a Supplier as one of the constructor arguments that can be used to populate the pool (instead of having an abstract createObject method) (also, you may want to actually implement the Supplier interface yourself – it may be useful for others – consider it later).
Instead of having a dedicated system for “monitoring” the pool size with a scheduled thread system, you should rather have hooks on the borrow/return side of things to trigger asynchronous pool modifications/alterations.
Finally, you should implement the queue inside the abstract class completely, and not expose that to higher-up implementations of the class.
Restructured
Having said all those things, here’s a “different” way of implementing the pool…
First up, the basic interface if you want to separate it out in its use contexts:
public interface Pool<T> {
T borrowItem() throws PoolDepletionException, InterruptedException;
void returnItem(T item);
}
Note that I ended up with borrowItem and returnItem because return is a reserved word. Hmmm… I think I will re-recommend implementing the Supplier interface, and rename borrowItem to be get, and returnItem to be put….
Then, an abstract implementation that does everything except handle what to do in the event that the pool is depleted.
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
public abstract class AbstractPool<T> implements Pool<T> {
private final BlockingQueue<T> queue;
private final Supplier<T> supplier;
private final long maxWait;
private final int minIdle;
private final int maxIdle;
protected AbstractPool(Supplier<T> supplier) {
this(supplier, 1, Integer.MAX_VALUE, 0);
}
protected AbstractPool(Supplier<T> supplier, int minIdle, int maxIdle, long maxWait) {
if (minIdle < 0 || minIdle >= maxIdle) {
throw new IllegalArgumentException("minIdle must be non-negative, and also strictly less than maxIdle");
}
this.supplier = supplier;
this.minIdle = minIdle;
this.maxIdle = maxIdle;
this.queue = new LinkedBlockingQueue<>(maxIdle);
this.maxWait = maxWait;
// Set up a feed of input values.
if (this.minIdle > 0) {
lazyAdd(this.minIdle);
}
}
public int getMinIdle() {
return minIdle;
}
public int getMaxIdle() {
return maxIdle;
}
public int getCurrentIdle() {
return queue.size();
}
/**
* lazyAdd will start a background thread to add instances to the queue (if
* possible). The cost of creating a thread is assumed to be insignificant
* compared to the cost of creating the instance.
*
* @param count
* the number of items to add to the pool
*/
private void lazyAdd(final int count) {
Runnable toRun = () -> {
for (int i = 0; i < count; i++) {
if (!queue.offer(supplier.get())) {
return;
}
}
};
Thread t = new Thread(toRun);
t.setDaemon(true);
t.start();
}
@Override
public void returnItem(T item) {
if (item == null) {
return;
}
queue.offer(item);
}
@Override
public T borrowItem() throws PoolDepletionException, InterruptedException {
T got = queue.poll();
// regardless of other things, add a lazy instance if the current state
// is too low.
if (queue.size() < minIdle) {
lazyAdd(1);
}
if (got != null) {
return got;
}
if (maxWait > 0) {
// Try again but block until there's something available.
// The lazy add above may be the instance we pull out!
// It is also possible that some other thread may steal the one we
// added!
got = queue.poll(maxWait, TimeUnit.MILLISECONDS);
}
if (got != null) {
return got;
}
return handleDepleted(supplier);
}
@Override
public String toString() {
return String.format("Pool with %d items and min %d, max %d wait %dms", queue.size(), minIdle, maxIdle,
maxWait);
}
/**
* What to do if there's nothing to return within the expected time. options
* are to create an instance manually, or alternatively throw an exception.
* returning null would be a bad idea, though.
*
* @param supplier
* the supplier that feeds this pool - can be used to get a
* guaranteed, but slow instance.
* @return the value for the client to use
* @throws PoolDepletionException
* if that is how the pool instance should respond.
*/
protected abstract T handleDepleted(Supplier<T> supplier) throws PoolDepletionException;
}
Note that the code above handles all the basic situations you describe – it has the ability to feed items in to the queue if the idle count is low, but that can be “turned off” by setting the minIdle count to 0.
In addition, it restricts access to the underlying queue by fully encapsulating it, and not exposing it.
The implementing classes only need to worry about how to handle a situation where an instance is not available “in time”, and it has 2 choices – either give it a new instance, or throw an exception. For convenience, the actual supplier is passed in as well, so it can use that if needed.
If you are curious, I set up a “stupid” little test system to run the code. By messing around with the different thread sleep times, you can trigger different behaviour:
import java.util.function.Supplier;
public class SlowStringPool extends AbstractPool<String> {
public SlowStringPool() {
super(buildSupplier(), 10, 20, 99);
}
private static Supplier<String> buildSupplier() {
return () -> {
try {
Thread.sleep(100);
} catch (InterruptedException e) {
// nothing
}
return String.valueOf(System.currentTimeMillis());
};
}
public SlowStringPool(Supplier<String> supplier) {
super(supplier);
}
@Override
protected String handleDepleted(Supplier<String> supplier) throws PoolDepletionException {
throw new PoolDepletionException();
}
public static void main(String[] args) throws InterruptedException, PoolDepletionException {
SlowStringPool mypool = new SlowStringPool();
Thread.sleep(2000);
int i = 0;
while (true) {
long start = System.nanoTime();
String s = mypool.borrowItem();
long durn = System.nanoTime() - start;
System.out.printf("%4d -> %s in %10.3fms -> %sn", i++, s, durn / 1000000.0, mypool.toString());
if (i % 2 == 0) {
mypool.returnItem(s);
}
}
}
}