Java Schedulable CompletionService

This week I wanted to knock-up a tool to test the performance of our Monte Carlo grid with a set of more complex algorithms so it seemed like a good idea to start by replicating flows from production. Each incoming trade is timestamped on receipt and thus could be replayed into the test grid at exactly the same rate. The Java language has an excellent set of tools for multi-threading and quite complex problems can be solved with relative ease given a solid understanding of classic threading models. However this time I noticed that there had been an omission of some importance from the concurrency package, while there is a ExecutorCompletionService class there is no complimentry Scheduled ExecutorCompletionService to complement the ScheduledThreadPoolExecutor and while as in most languages it is easy enough to “roll your own” it was not quite as straightforward as I first thought but after a little cogitating I came up with a straightforward solution and thought I’d share.

First of all you need to wrap the data you want to carry within the schedule in a Callable and in turn wrap that in a Future object prior to scheduling within the ExecutorCompletionService. The subtle bit is passing in a reference to the external CompletionService during construction of the ScheduledJob object and then instantiating a WeakReference to keep a reference to it for the call() override; this makes sure that our data items can be garbage collected later-on when they are no longer required.

Below is an example of how to do this with classes Job and ScheduledJob which have been specialised to carry a data payload of type Map<String, Object> which is general purpose and handy for most database uses but you can use anything else you like.

Definitions should be provided for the ScheduledExecutorService and CompletionService like this, remembering that the ExecutorService should be initialised with a thread-pool size suitable for your solution.

All that then remains is to instantiate your data payload, wrap it in a Job which is wrapped in a ScheduledJob and schedule it within the ScheduledExecutorService with your desired delay. Retrieval (as shown below) is a matter of calling thread blocking function take() on the CompletionService, which returns the next chronologically available Future object. Obtaining our original data is achived through a call to get() and we’re done.

You can download the example Eclipse project here and experiment yourself.

For further reading I recommend Java Concurrency in Practice by Brian Goetz.