Executor Framework- Understanding the basics (Part 2)
In my previous post, I had discussed about the basics of Executor framework. In this post, I am going to discuss about BlockingQueues. BlockingQueue is an integral part of an Executor. BlockingQueue is used for storing the tasks when all the available threads are busy executing tasks and a new thread cannot be created. A developer should consider the following things while choosing a particular BlockingQueue.
Size of the queue : What should be size of the queue? Should the queue have a fixed size or should the queue be unbounded?
Order of execution of the task : What should be the order of execution of the tasks stored in the queue? Should a task which was submitted first be executed first or should the tasks be picked up based on their priority?
The executor framework provides different kinds of BlockingQueues. Lets see the different types of queues available in the framework and how do they differ based on the above parameters.
LinkedBlockingQueue
Let’s start with LinkedBlockingQueue which I had used in my previous post. The size of the queue is dynamic- i.e. it will be created with an initial size and then as the tasks keeps on adding to the queue, its size will be increased. Theoretically it can add up-to Integer.MAX_VALUE elements. Optionally the user can give a capacity to the queue using LinkedBlockingQueue(int capacity) constructor so that the queue doesn’t expand to more elements than are required. The queue works in FIFO approach. So the tasks, which are submitted first are executed first.
ArrayBlockingQueue
As the name suggests, the queue internally uses an array of fixed size to hold the tasks. While creating an instance of ArrayBlockingQueue, we need to specify the size of the queue. The queue works in FIFO fashion- i.e a task which is submitted first will be executed first. But if the queue is full and a new task is submitted, then the queue will reject that task.
Now in this code, let’s replace LinkedBlockingQueue with ArrayBlockingQueue which has a size of 2 elements.
private final ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 2, 0,
TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(2));What happens when we run the above code?
When the 1st task is picked up, a new thread will be created for its execution. Similarly when the 2nd task is picked up, a new thread will be created. Now since both the threads are busy executing tasks, any new incoming tasks will be stored in the queue. Therefore 3rd and 4th tasks will be stored in the queue. But after storing 2 tasks , the queue becomes full. When the 5th task is submitted, the queue is already full and cannot accept any incoming task. The executor will throw RejectedExecutionException. Similarly 6th task will be rejected. So only the first 4 tasks will be executed and the remaining 2 will be rejected.
PriorityBlockingQueue
This is a special type of queue in which the executor picks up a task from the queue based on its priority and not its order. While creating a task, we need to assign a priority to every task.
The executor uses Comparable interface to find the task with higher priority. In the code below, we create a PriorityRunnable which extends Runnable and Comparable interfaces. Through setPriority() method we set the priority for the runnable.
If the Runnable submitted to the executor does not implement Comparable, then it will throw the following exception.
PriorityRunnable cannot be cast to java.lang.Comparable
Now let’s modify our code of StringPrinter.java . In this code, we are passing a ‘length’ variable to the getRunnable(int length) method. Now we create a new method getPriorityRunnable(int length). This method will create a PriorityRunnable and assign a HIGH priority to it if the length passed is even and LOW priority if the length passed is odd.
What happens when we run the code?
The Runnables which were created with even length will be executed first and then the odd length Runnables will be executed. But if two Runnables have the same priority, then the executor can pick any one of them. The order of execution of the Runnables with same priority is not certain. The executor can pick up the tasks with same priority in any order.
But be cautious in assigning the priority to the tasks. The queue can result in starvation of certain tasks as there is a chance that the tasks with low priority will never be picked.
LinkedBlockingDeque
Similar to LinkedBlockingQueue, this queue is also dynamic in size and optionally unbounded. But unlike LinkedBlockingQueue, where the tasks are always appended to the end of the queue, here we can append a task at the beginning of the queue also. Basically we can access the queue from both the ends. We can insert a task either to the starting or the end of the queue. The executor can pick up a task for execution either from the starting or the end of the queue.
//Get the underlying queue from the executor
LinkedBlockingDeque queue = (LinkedBlockingDeque) executor.getQueue();//Insert the task to the beginning of the queue queue.addFirst(runnable);
DelayQueue
This is also a special type of queue, in which every task is created with a delay period . The executor will not execute a task before its delay has been expired. The executor uses Delayed interface to compare the two tasks for their delay period. In the below code, we create a DelayedRunnable which extends Runnable and Delayed interfaces. Through setDelay(int delay) method we set the delay for the runnable.
If the Runnable submitted to the executor does not implement Delayed, then it will throw the following exception.
DelayedRunnable cannot be cast to java.util.concurrent.Delayed
Now let’s modify our code of StringPrinter.java . Till now we were passing a ‘length’ variable to the getRunnable(int length) method. Now we create a new method getDelayedRunnable(int length). This method will create a DelayedRunnable and assign a delay of 3000 milliseconds if the length is odd and 6000 milliseconds if the length is even.
What happens when we run the code?
The Runnables which were created with odd length will be executed first because they have shorter delay period i.e. 3000 milliseconds . After that the Runnables which were created by passing by even length will be executed. But if two Runnables have the same delay period, then the executor can pick any one of them. The order of execution of the Runnables with same delay period is not certain.
Similar to PriorityBlockingQueue, this queue can result in starvation of certain tasks as there is a chance that the tasks with higher delay period will never be picked.
SynchronousQueue
This is a special kind of queue and even though it is a queue, it does not hold any elements (tasks). The function of this queue is to receive the tasks and pass them for execution but not hold them. If the queue receives a task, and if a worker thread is present, it will pass on that task for execution to that thread, otherwise it will reject the task. Optionally it can wait for some time till the thread becomes available for executing the task before rejecting the task.
What happens when the queue is full?
Till now, the task of the executor seemed simple enough — Execute a task on the thread if the thread is available or put the task on the queue for later execution. But what happens when a task is submitted to the queue and the queue is full? Simple. The queue will reject the task. But this can be handled. While adding a task to the queue, you can specify to wait for a certain time if the queue is full. Only after the specified duration has been passed and the queue is still not empty, the task will be rejected.
You can also pass a RejectedExecutionHandler while creating an executor which gets back the rejected tasks in its rejectedExecution(Runnable r, ThreadPoolExecutor executor) method. From here, you can decide what to do with the rejected tasks. The below code snippet tries to execute the rejected task after 3 seconds.
Thanks for reading the post guys.
