October 9, 2015 / Last updated : April 22, 2023 Drona Java

Tracking thread pool in Java

This article presents an implementation of a tracking thread pool in Java.

A tracking thread pool is a usual thread pool that records the task execution statistics like total number of tasks executed, average execution time of tasks and total execution time of all tasks. The default thread pools created by Executors class in Java does not track the execution stats leaving you no option but to implement your own thread pool that does the job.

In this article, I present an implementation of a tracking thread pool that tracks the total execution time, average execution time and the total tasks executed and provides methods to query those parameters. This is quite a powerful feature to have in a thread pool.

You can go a step further and expose these thread statistics through a JMX bean that can be queried and monitored through monitoring tools whenever desired.

Here is the implementation.

TrackingThreadPool.java


/*
 * TrackingThreadPool - A thread pool that tracks task execution statistics
 *
 * Copyright (c) 2015-2016, Wilddiary.com
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without modification, are permitted provided that the
 * following conditions are met:
 *
 *    1. Redistributions of source code must retain the above copyright notice, this list of conditions and the
 *       following disclaimer.
 *    2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the
 *       following disclaimer in the documentation and/or other materials provided with the distribution.
 *    3. The name of the author may not be used to endorse or promote products derived from this software without
 *       specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
 * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
 * THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
 * BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 */

package com.wilddiary.concurrency;

import java.util.Collections;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class TrackingThreadPool extends ThreadPoolExecutor {
    
    // holds in-progress tasks
    private final Map<Runnable, Boolean> inProgress = new ConcurrentHashMap<Runnable,Boolean>(); 
	
    // holds startime of the currently executing task. 
    // Each thread has its own copy of this variable
    private final ThreadLocal<Long> startTime = new ThreadLocal<Long>(); 
   
    private volatile long totalTime; // in ms
    
    private volatile long totalTasks;

    // Constructor
    public TrackingThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime,
       TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory factory) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, factory);
    }
    
    // Called by each thread before executing a task
    // The executing thread is t and the executed task is r
    protected void beforeExecute(Thread t, Runnable r) {
       
        super.beforeExecute(t, r);
        inProgress.put(r, Boolean.TRUE);
       
        // record the start time
        startTime.set(new Long(System.currentTimeMillis())); 
    }

    // Called by each thread after executing the task
    // The executing thread is t and the executed task is r
    protected void afterExecute(Runnable r, Throwable t) {
    	
    	// calculate the time taken by the task to execute
        long time = System.currentTimeMillis() - startTime.get().longValue();
        
        synchronized (this) { 
        	// update the stats of the pool
            totalTime += time;
            ++totalTasks;
        } 
        
        inProgress.remove(r);
        super.afterExecute(r, t);
    }

    /**
     * @return - returns the currently executing tasks
     */    
    public Set<Runnable> getInProgressTasks() {
        return Collections.unmodifiableSet(inProgress.keySet());
    }

    /**
     * @return - returns the total tasks executed till that moment
     */
    public synchronized long getTotalTasks() {
        return totalTasks;
    }

    /**
     * @return - returns the average execution time of tasks in ms
     */
    public synchronized double getAverageExecutionTime() {
        return (totalTasks == 0) ? 0 : totalTime / (double) totalTasks;
    }
    
    /**
     * @return - returns the total execution time of tasks in ms
     */
    public synchronized long getTotalExecutionTime() {
        return totalTime;
    }
}

100

101

102

103

104

105

106

107

108

109

110

111

112

* TrackingThreadPool - A thread pool that tracks task execution statistics

* Redistribution and use in source and binary forms, with or without modification, are permitted provided that the

* following conditions are met:

* 1. Redistributions of source code must retain the above copyright notice, this list of conditions and the

* following disclaimer.

* 2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the

* following disclaimer in the documentation and/or other materials provided with the distribution.

* 3. The name of the author may not be used to endorse or promote products derived from this software without

* specific prior written permission.

* THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED

* TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL

* THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,

* BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS

* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT

* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE

* POSSIBILITY OF SUCH DAMAGE.

package com.wilddiary.concurrency;

import java.util.Collections;

import java.util.Map;

import java.util.Set;

import java.util.concurrent.BlockingQueue;

import java.util.concurrent.ConcurrentHashMap;

import java.util.concurrent.ThreadFactory;

import java.util.concurrent.ThreadPoolExecutor;

import java.util.concurrent.TimeUnit;

public class TrackingThreadPool extends ThreadPoolExecutor {

// holds in-progress tasks

private final Map<Runnable, Boolean> inProgress = new ConcurrentHashMap<Runnable,Boolean>();

// holds startime of the currently executing task.

// Each thread has its own copy of this variable

private final ThreadLocal<Long> startTime = new ThreadLocal<Long>();

private volatile long totalTime; // in ms

private volatile long totalTasks;

// Constructor

public TrackingThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime,

TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory factory) {

super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, factory);

}

// Called by each thread before executing a task

// The executing thread is t and the executed task is r

protected void beforeExecute(Thread t, Runnable r) {

super.beforeExecute(t, r);

inProgress.put(r, Boolean.TRUE);

// record the start time

startTime.set(new Long(System.currentTimeMillis()));

}

// Called by each thread after executing the task

// The executing thread is t and the executed task is r

protected void afterExecute(Runnable r, Throwable t) {

// calculate the time taken by the task to execute

long time = System.currentTimeMillis() - startTime.get().longValue();

synchronized (this) {

// update the stats of the pool

totalTime += time;

++totalTasks;

}

inProgress.remove(r);

super.afterExecute(r, t);

}

/**

* @return - returns the currently executing tasks

public Set<Runnable> getInProgressTasks() {

return Collections.unmodifiableSet(inProgress.keySet());

}

/**

* @return - returns the total tasks executed till that moment

public synchronized long getTotalTasks() {

return totalTasks;

}

/**

* @return - returns the average execution time of tasks in ms

public synchronized double getAverageExecutionTime() {

return (totalTasks == 0) ? 0 : totalTime / (double) totalTasks;

}

/**

* @return - returns the total execution time of tasks in ms

public synchronized long getTotalExecutionTime() {

return totalTime;

}

TrackingThreadPool tester


package com.wilddiary.concurrency;

import java.util.Random;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

/**
 * Tracking thread pool tester
 * 
 * @author Drona
 */
public class TrackingThreadPoolTest {
    
    public static void main(String[] args) throws InterruptedException {
        
        // create the tracking thread pool with 5 threads
        final TrackingThreadPool pool = newTrackingThreadPool(5);
        
        // submit some tasks for execution
        for (int i = 0; i < 50; i++) {
            pool.submit(new RandomNumberPrinterTask());
        }
        
        // shutdown the pool
        pool.shutdown();
        
        // await shutdown
        while (!pool.isTerminated()){
            pool.awaitTermination(2, TimeUnit.SECONDS);
        }
        
        // print the execution stats
        System.out.println("----------------------------------");
        System.out.println("Total tasks executed = " + pool.getTotalTasks());
        System.out.println("Average execution time = " + pool.getAverageExecutionTime() + " ms");
        System.out.println("Total execution time = " + pool.getTotalExecutionTime() + " ms");
        System.out.println("----------------------------------");
    }
    
    /**
     * Factory method to create the tracking thread pool with the given number
     * of threads
     * 
     * @param numOfThreads
     * @return
     */
    private static TrackingThreadPool newTrackingThreadPool(int numOfThreads){
        return new TrackingThreadPool(numOfThreads, numOfThreads, 0L,
                TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), new ThreadFactory() {

            @Override
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r);
                t.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
                    @Override
                    public void uncaughtException(Thread t, Throwable e) {
                        System.err.println("Thread terminated due to exception - " + e.getMessage());
                    }
                });
                return t;
            }
        });
    }

    /**
     * Random number printer task.
     * 
     * @author Drona
     *
     */
    static class RandomNumberPrinterTask implements Runnable{
        
        @Override
        public void run() {
            int random = new Random().nextInt(100);
            
            System.out.println(Thread.currentThread().getName() + " - " +  random);
            
            try {
                Thread.sleep(random);
            } catch (InterruptedException e) {
                System.err.println("Thread interrupted");
            }
        }    
    }
}

package com.wilddiary.concurrency;

import java.util.Random;

import java.util.concurrent.LinkedBlockingQueue;

import java.util.concurrent.ThreadFactory;

import java.util.concurrent.TimeUnit;

/**

* Tracking thread pool tester

* @author Drona

public class TrackingThreadPoolTest {

public static void main(String[] args) throws InterruptedException {

// create the tracking thread pool with 5 threads

final TrackingThreadPool pool = newTrackingThreadPool(5);

// submit some tasks for execution

for (int i = 0; i < 50; i++) {

pool.submit(new RandomNumberPrinterTask());

}

// shutdown the pool

pool.shutdown();

// await shutdown

while (!pool.isTerminated()){

pool.awaitTermination(2, TimeUnit.SECONDS);

}

// print the execution stats

System.out.println("----------------------------------");

System.out.println("Total tasks executed = " + pool.getTotalTasks());

System.out.println("Average execution time = " + pool.getAverageExecutionTime() + " ms");

System.out.println("Total execution time = " + pool.getTotalExecutionTime() + " ms");

System.out.println("----------------------------------");

}

/**

* Factory method to create the tracking thread pool with the given number

* of threads

* @param numOfThreads

* @return

private static TrackingThreadPool newTrackingThreadPool(int numOfThreads){

return new TrackingThreadPool(numOfThreads, numOfThreads, 0L,

TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), new ThreadFactory() {

@Override

public Thread newThread(Runnable r) {

Thread t = new Thread(r);

t.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {

@Override

public void uncaughtException(Thread t, Throwable e) {

System.err.println("Thread terminated due to exception - " + e.getMessage());

}

});

return t;

}

});

}

/**

* Random number printer task.

* @author Drona

static class RandomNumberPrinterTask implements Runnable{

@Override

public void run() {

int random = new Random().nextInt(100);

System.out.println(Thread.currentThread().getName() + " - " + random);

try {

Thread.sleep(random);

} catch (InterruptedException e) {

System.err.println("Thread interrupted");

}

Output


Thread-2 - 90
Thread-3 - 87
Thread-1 - 54
Thread-4 - 43
Thread-0 - 49
Thread-4 - 87
Thread-0 - 72
Thread-1 - 10
Thread-1 - 53
Thread-3 - 44
Thread-2 - 98
Thread-1 - 12
Thread-0 - 97
Thread-1 - 77
Thread-4 - 44
Thread-3 - 53
Thread-4 - 0
Thread-4 - 38
Thread-2 - 70
Thread-3 - 53
Thread-1 - 66
Thread-0 - 38
Thread-4 - 72
Thread-3 - 78
Thread-2 - 72
Thread-0 - 15
Thread-0 - 11
Thread-1 - 8
Thread-0 - 68
Thread-1 - 43
Thread-4 - 24
Thread-4 - 78
Thread-3 - 5
Thread-3 - 36
Thread-2 - 85
Thread-1 - 67
Thread-0 - 51
Thread-3 - 19
Thread-3 - 40
Thread-4 - 16
Thread-1 - 34
Thread-0 - 56
Thread-4 - 52
Thread-2 - 90
Thread-3 - 15
Thread-1 - 46
Thread-3 - 60
Thread-0 - 97
Thread-4 - 19
Thread-1 - 95
----------------------------------
Total tasks executed = 50
Average execution time = 54.72 ms
Total execution time = 2736 ms
----------------------------------

Thread-2 - 90

Thread-3 - 87

Thread-1 - 54

Thread-4 - 43

Thread-0 - 49

Thread-4 - 87

Thread-0 - 72

Thread-1 - 10

Thread-1 - 53

Thread-3 - 44

Thread-2 - 98

Thread-1 - 12

Thread-0 - 97

Thread-1 - 77

Thread-4 - 44

Thread-3 - 53

Thread-4 - 0

Thread-4 - 38

Thread-2 - 70

Thread-3 - 53

Thread-1 - 66

Thread-0 - 38

Thread-4 - 72

Thread-3 - 78

Thread-2 - 72

Thread-0 - 15

Thread-0 - 11

Thread-1 - 8

Thread-0 - 68

Thread-1 - 43

Thread-4 - 24

Thread-4 - 78

Thread-3 - 5

Thread-3 - 36

Thread-2 - 85

Thread-1 - 67

Thread-0 - 51

Thread-3 - 19

Thread-3 - 40

Thread-4 - 16

Thread-1 - 34

Thread-0 - 56

Thread-4 - 52

Thread-2 - 90

Thread-3 - 15

Thread-1 - 46

Thread-3 - 60

Thread-0 - 97

Thread-4 - 19

Thread-1 - 95

----------------------------------

Total tasks executed = 50

Average execution time = 54.72 ms

Total execution time = 2736 ms

----------------------------------

Categories: Java

Tags: thread-pool

0 0 votes

Article Rating

0 Comments

Inline Feedbacks

View all comments

Tracking thread pool in Java

TrackingThreadPool.java

TrackingThreadPool tester

Output

How to create a deadlock in Java

Java String to Int