Multi-Threading In Java

Performance makes or breaks an application and in this Digital age of short attention spans, companies are looking for high-performance applications -Multithreading is a powerful weapon to achieve this target.

What is Multithreading

We have all heard the phrase Divide and conquer, well Multithreading is basically doing the same, by dividing a task into multiple tasks, and executing each task Independently. Java application takes the help of threads to execute the tasks in parallel, and it provides a list of Classes with various varieties of choices to achieve this goal. In this blog, we will be going through the different ways to achieve Multithreading.

Threads

A thread is a thread of execution in a program. The Java Virtual Machine allows an application to have multiple threads of execution running concurrently. Every thread has a priority. Threads with higher priority are executed in preference to threads with lower priority. Each thread may or may not also be marked as a daemon. When code running in some thread creates a new Thread object, the new thread has its priority initially set equal to the priority of the creating thread, and is a daemon thread if and only if the creating thread is a daemon.

When a Java Virtual Machine starts up, there is usually a single non-daemon thread (which typically calls the method named main of some designated class). The Java Virtual Machine continues to execute threads until either of the following occurs:

• The exit method of class Runtime has been called and the security manager has permitted the exit operation to take place.
• All threads that are not daemon threads have died, either by returning from the call to the run method or by throwing an exception that propagates beyond the run method.

There are two ways to create a new thread of execution. One is to declare a class to be a subclass of Thread. This subclass should override the run method of class Thread. An instance of the subclass can then be allocated and started.

Extending Thread Class

     class CustomThread extends Thread {

         public void run() {
             // Execute your task
              . . .
         }
     }

The following code would then create a thread and start it running:

CustomThread p = new CustomThread();
new Thread(p).start();

Implementing Runnable

The other way to create a thread is to declare a class that implements the Runnable interface. That class then implements the run method. An instance of the class can then be allocated, passed as an argument when creating, and started

public class CustomRunnable implements Runnable{

    @Override
    public void run() {
        // .. Do your stuff
    }

    public static void main(String[] args) {
        // Metod 1 
        CustomRunnable p = new CustomRunnable();
        new Thread(p).start();
        // Metod 2
        Runnable runnable = () -> System.out.println("Hello");
        new Thread(runnable).start();
    }
}

Executor Framework

Executor service, Executors, and its different implementations are part of java.util.concurrent package, they provide classes that help us to execute tasks parallelly. If we are going to do this manually, then we need to implement things like creating threads, maintaining state, and maintaining thread results and all this generates a lot of boiler code.

Runnable Vs Callable

ExecutorService is an interface and its implementations can execute a Runnable or Callable class in an asynchronous way.
The difference between Runnable and Callable can be seen in its implementation. Callable throws an Exception and return an Object whereas Runnable does not do any of this and is just executing the supplied instruction.

// Runnable Implementation
public interface Runnable {
    public void run();
}
// Callable Implementation
public interface Callable{
    public Object call() throws Exception;
}

ThreadPoolExecutor Examples

Example1 – FixedThreadPool

Creates a thread pool that reuses a fixed number of threads operating off a shared unbounded queue.

        ExecutorService fixedExecutorService = Executors.newFixedThreadPool(10);
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("uuuu-MM-dd'T'HH:mm:ss.SSS");
        Runnable fixedExecutorServiceRunnable = () -> {
                System.out.println("Fixed Executor Time -- " + LocalDateTime.now().format(formatter));
        } ;
        Future obj = fixedExecutorService.submit(fixedExecutorServiceRunnable);
        System.out.println(obj.get()); //returns null if the task has finished correctly.
        fixedExecutorService.shutdown();

Example2 – ScheduledThreadPool

Creates a thread pool that can schedule commands to run after a given delay, or to execute periodically.

        Callable<String> scheduledCallableTask = () -> "Scheduled Callable Task's execution";
        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(10);
        Future<String> resultFuture =
                executorService.schedule(scheduledCallableTask, 2, TimeUnit.SECONDS);
        System.out.println(resultFuture.get());
        executorService.shutdown();

Example3 – CachedThreadPool

Creates a thread pool that creates new threads as needed, but will reuse previously constructed threads when they are available, and uses the provided ThreadFactory to create new threads when needed.

        // No of threads is restricted to resources and then disposed with 60s of inactivity
        ThreadPoolExecutor cachedExecutor =
                (ThreadPoolExecutor) Executors.newCachedThreadPool();
        Future<String> futureCachedThreadPool = cachedExecutor.submit(() -> "CachedExecutor Task's execution");
        System.out.println(futureCachedThreadPool.get());
        cachedExecutor.shutdown();

Example4 – WorkStealingPool

Fork Join is an implementation of ExecuterService. The main difference is that this implementation creates DEQUE worker pool. Where task is inserted from oneside but withdrawn from any side. It means if you have created new ForkJoinPool() it will look for the available CPU and create that many worker thread. It then distribute the load evenly across each thread. But if one thread is working slowly and others are fast, they will pick the task from the slow thread

        ExecutorService workStealingPool = Executors.newWorkStealingPool();
        Future<String> futureWorkStealing = workStealingPool.submit(() -> "WorkStealingPool Callable Task's execution");
        System.out.println(futureWorkStealing.get());
        workStealingPool.shutdown();

Example5 – ThreadPoolExecutor

ThreadPoolExecutor gives us more control on how the threads needs to be executed.

        ExecutorService threadPoolExecutorService =
                new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
                        new LinkedBlockingQueue<Runnable>());
        // corePoolSize : The corePoolSize parameter is the number of core threads that will be instantiated and kept in the pool.
        // maximumPoolSize : When a new task comes in, if all core threads are busy and the internal queue is full, the pool is allowed to grow up to maximumPoolSize.
        // keepAliveTime : The keepAliveTime parameter is the interval of time for which the excessive threads (instantiated in excess of the corePoolSize) are allowed
        // to exist in the idle state
        Callable<String> callableTask = () -> "Callable Task's execution";
        Future<String> future =
                threadPoolExecutorService.submit(callableTask);
        System.out.println(future.get());
        threadPoolExecutorService.shutdown();

Example6 – Java8 libraries FixedThreadPool to execute tasks parallelly

        Executor fixedThreadPool = Executors.newFixedThreadPool(10);
        List<CompletableFuture<CustomClass>> futures =
                              objects.stream()
                              .map( objectId -> CompletableFuture.supplyAsync(() -> callYourMethod(objectId), fixedThreadPool))
                              .collect(Collectors.toList());
        List<CustomClass> customList =  futures.stream()
                                               .map(CompletableFuture::join)
                                               .filter(Objects::nonNull)
                                               .collect(Collectors.toList());

Example7 – DelegatingSecurityContextExecutor

This is used to pass the security context to the child threads.

        Executor delegatedExecutor = Executors.newFixedThreadPool(10);
        Executor delegatingExecutor = new DelegatingSecurityContextExecutor(delegatedExecutor);
        Runnable singleThreadExecutorRunnable = () ->  System.out.println("DelegatingExecutor Executor Time -- " + LocalDateTime.now());
        delegatingExecutor.execute(singleThreadExecutorRunnable);