CustomThreadPoolExecutor in Java Executor Framework

0 1 5 minutes read

Executors Manage thread execution. At the top of the executor, hierarchy is the Executor interface, which is used to initiate a thread. ExecutorService Extends Executor and provides methods that manage execution. There are three implementations of ExecutorService: ThreadPoolExecutor, ScheduledThreadPoolExecutor, and ForkJoinPool. java.util.concurrent also defines the Executors utility class, which includes some static methods that simplify the creation of various executors. Related to executors are the Future and Callable interfaces. A Future contains a value that is returned by a thread after it executes. Thus, its value becomes defined “in the future,” when the thread terminates. Callable defines a thread that returns a value. In this article, we are going to learn about Custom ThreadPoolExecutor in java.

First, let us discuss two concepts been aggressively used here namely thread pool and blocking queue.

ThreadPool is a container in which contains some numbers of threads. These threads are given some tasks. When one thread completes its task next task is given to it. While working in a multi-threading environment it’s not practical to create new individual threads for each new task, because creating a new thread is overhead for the operating system.
A blocking queue is a queue that blocks when you try to dequeue from it and the queue is empty, If you try to enqueue items to t and the queue is already full. All operations in the blocking queue are thread-safe.

Also, the important specific methods that are to be implemented are as follows:

Method 1: execute()

This method is contained in the Executor interface. This function executes the given task at some time in the future. It returns nothing hence the return type of this method is void.

Method 2: myNewFixedThreadPool()

This is a factory method of Executors class. It is used to create a fixed number of threads in the thread pool.

Parameter: int number of threads
Return type: ExecutorService

Procedure:

Create an interface in which we will create a execute method. This method will execute the task given to it.
In the code above generated, we have implemented a runnable interface. We are printing the current name of the thread with a delay of 1000 milliseconds. These are the tasks which we are going to execute.
MyExecutor class provides a static method myNewFixedThreadPool in which we will pass the number of threads we want to create. This method tells the thread pool that how many threads are going to be there in the thread pool. These threads will execute tasks till all tasks get completed.
This is the custom thread pool class. This class is the heart of the whole mechanism. It uses two important concepts LinkedBlockingQueue and Execution class. The execution class is explained further. This class receives the thread count from the myNewFixedThreadPool method. All the tasks we submit are stored in the queue. All the threads will fetch the tasks from the queue. We submit the task by using the execute method of the MyExecuorService.
Execution class performs the very important task of adding creating the number of threads that we want in our thread pool. This class is where we are defining how to fetch the task from LinkedBlockingQueue.
Finally, in this class, we are gathering all the pieces together and our custom thread pool is ready.

Implementation: Here we are passing some threads as 3. The number of tasks is 20 and executing them by using execute method.

Java

// Java Program to illustrate Concept of
// CustomThreadPoolExecutor Executor Framework
 
// Importing LinkedBlockingQueue class from java.util
// package
import java.util.concurrent.LinkedBlockingQueue;
 
// Interface
// Custom interface for which contains execute method
interface MyExecutorService {
 
    // Method
    void execute(Runnable r);
}
 
// Class 1
// Helper class
class MyExecutors {
 
    // Member variables of this class
    int capacity;
 
    // Passing the number of threads that
    // will be in the thread pool
    static MyExecutorService
    myNewFixedThreadPool(int capacity)
    {
 
        return new MyThreadPool(capacity);
    }
}
 
// Class 2
// Helper class extending to MyExecutorService interface
class MyThreadPool implements MyExecutorService {
 
    // Member variables of this class
    static int capacity;
    static int currentCapacity;
 
    // Creating object of LinkedBlockingQueue class
    // Declaring object of type Runnable
    static LinkedBlockingQueue<Runnable>
        linkedTaskBlockingQueue;
 
    // Member variables of this class
    Execution e;
 
    // Method 1
    public MyThreadPool(int capacity)
    {
 
        // Member variables of this class
 
        // this keyword refers to current instance itself
        this.capacity = capacity;
        currentCapacity = 0;
 
        // Creating a linked blocking queue which will block
        // if its empty
        // and it will perform thread safe operation.
        linkedTaskBlockingQueue
            = new LinkedBlockingQueue<Runnable>();
 
        // Creating the object of execution class
        e = new Execution();
    }
 
    // Method 2
    // @Override
    public void execute(Runnable r)
    {
 
        // Declaring and adding tasks to
        // blocking queue using add() method
        linkedTaskBlockingQueue.add(r);
 
        // executeMyMethod() method of Execution class
        // which will execute the tasks
        e.executeMyMethod();
    }
}
 
// Class 3
// Helper class extending Runnable interface
class Execution implements Runnable {
 
    // Method 1 of  this class
    void executeMyMethod()
    {
 
        // At start the current capacity will be 0
        // The another capacity is the number of threads we
        // want to create so we will increase the current
        // capacity count after creating each thread it
        // means that we will create the threads if the
        // current capacity is less than capacity passed by
        // us i.e number of threads we want to create.
 
        // In this case 3 threads will get created
        if (MyThreadPool.currentCapacity
            < MyThreadPool.capacity) {
            MyThreadPool.currentCapacity++;
 
            // Creating object of Thread class
            Thread t = new Thread(new Execution());
 
            // Starting the thread
            t.start();
        }
    }
 
    // Method 2 of this class
    // @Override
    public void run()
    {
 
        // Till it is true
        while (true) {
 
            // Here we are fetching the tasks from the
            // linkedblocking queue
            // which we have submitted using execute method
            // and executing them
            if (MyThreadPool.linkedTaskBlockingQueue.size()
                != 0) {
                MyThreadPool.linkedTaskBlockingQueue.poll()
                    .run();
            }
        }
    }
}
 
// Class 4
// Helper class
// Here we are creating a simple task
// which is printing current thread name
class Mytask implements Runnable {
 
    // Method 1 of this class
    // @Override
    public void run()
    {
 
        // Try block to check for exceptions
        try {
 
            // Making thread to pause for a second
            // using sleep() method
            Thread.sleep(1000);
        }
 
        // Catch block to check for exceptions
        catch (InterruptedException e) {
 
            // Print the exception scaling ith line number
            // using printStackTrace() method
            e.printStackTrace();
        }
 
        // Print and display the current thread using
        // currentThread() method by getting thread name
        // using getName() method
        System.out.println(
            "Current Thread :-> "
            + Thread.currentThread().getName());
    }
}
 
// Class 5
// Main Class
public class ExecutorServiceCustom {
    // Main driver method
    public static void main(String[] args)
    {
        // Getting the object of MyExcutorService by using
        //  the factory method myNewFixedThreadPool
 
        // Passing number of threads as 3
        MyExecutorService service
            = MyExecutors.myNewFixedThreadPool(3);
 
        for (int i = 0; i < 20; i++) {
 
            // Creating 20 tasks and passing them to execute
            service.execute(new Mytask());
        }
 
        Runnable runnableTask = null;
    }
}

Output:

Current Thread :-> Thread-0
Current Thread :-> Thread-1
Current Thread :-> Thread-2
Current Thread :-> Thread-0
Current Thread :-> Thread-1
Current Thread :-> Thread-2
Current Thread :-> Thread-0
Current Thread :-> Thread-1
Current Thread :-> Thread-2
Current Thread :-> Thread-0
Current Thread :-> Thread-1
Current Thread :-> Thread-2
Current Thread :-> Thread-0
Current Thread :-> Thread-1
Current Thread :-> Thread-2
Current Thread :-> Thread-0
Current Thread :-> Thread-1
Current Thread :-> Thread-2
Current Thread :-> Thread-0
Current Thread :-> Thread-1

Note: In the above output, we have printed the thread name as defined in the runnable 20 times as we have submitted 20 tasks which is visually described through a video below