How to Develop and Monitor Thread Pool Services Using Spring

Eren Avsarogullari

Nov. 28, 14 · Interview

Likes (0)

Comment

Save

29.6K Views

Thread Pools are very important to execute synchronous & asynchronous processes. This article shows how to develop and monitor Thread Pool Services by using Spring. Creating Thread Pool has been explained via two alternative methods.

Used Technologies :

JDK 1.6.0_21
Spring 3.0.5
Maven 3.0.2

STEP 1 : CREATE MAVEN PROJECT

A maven project is created as below. (It can be created by using Maven or IDE Plug-in).

STEP 2 : LIBRARIES

Spring dependencies are added to Maven’ s pom.xml.

<!-- Spring 3 dependencies -->
<dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-core</artifactId>
    <version>${spring.version}</version>
</dependency>
<dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-context</artifactId>
    <version>${spring.version}</version>
</dependency>

For creating runnable-jar, below plugin can be used.

<plugin>
    <groupId>org.apache.maven.plugins</groupId>
    <artifactId>maven-shade-plugin</artifactId>
    <version>1.3.1</version>
    <executions>
        <execution>
            <phase>package</phase>
            <goals>
                <goal>shade</goal>
            </goals>
            <configuration>
                <transformers>
                    <transformer
                        implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                        <mainClass>com.otv.exe.Application</mainClass>
                    </transformer>
                    <transformer
                        implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
                        <resource>META-INF/spring.handlers</resource>
                    </transformer>
                    <transformer
                        implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
                        <resource>META-INF/spring.schemas</resource>
                    </transformer>
                </transformers>
            </configuration>
        </execution>
    </executions>
</plugin>

STEP 3 : CREATE TASK CLASS

A new TestTask Class is created by implementing Runnable Interface. This class shows to be executed tasks.

package com.otv.task;
 
import org.apache.log4j.Logger;
 
/**
 * @author onlinetechvision.com
 * @since 17 Oct 2011
 * @version 1.0.0
 *
 */
public class TestTask implements Runnable {
     
    private static Logger log = Logger.getLogger(TestTask.class);
    String taskName;
 
    public TestTask() {
    }
 
    public TestTask(String taskName) {
        this.taskName = taskName;
    }
 
    public void run() {
        try {
            log.debug(this.taskName + " : is started.");
            Thread.sleep(10000);
            log.debug(this.taskName + " : is completed.");
        } catch (InterruptedException e) {
            log.error(this.taskName + " : is not completed!");
            e.printStackTrace();
        }       
    }
 
    @Override
    public String toString() {
        return (getTaskName());
    }
 
    public String getTaskName() {
        return taskName;
    }
 
    public void setTaskName(String taskName) {
        this.taskName = taskName;
    }
}

STEP 4 : CREATE TestRejectedExecutionHandler CLASS

TestRejectedExecutionHandler Class is created by implementing RejectedExecutionHandler Interface. If there is no idle thread and queue overflows, tasks will be rejected. This class handles rejected tasks.

package com.otv.handler;
 
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
 
import org.apache.log4j.Logger;
 
/**
 * @author onlinetechvision.com
 * @since 17 Oct 2011
 * @version 1.0.0
 *
 */
public class TestRejectedExecutionHandler implements RejectedExecutionHandler {
     
    private static Logger log = Logger.getLogger(TestRejectedExecutionHandler.class);
     
    public void rejectedExecution(Runnable runnable, ThreadPoolExecutor executor) {
        log.debug(runnable.toString() + " : has been rejected");
    }
}

STEP 5 : CREATE ITestThreadPoolExecutorService INTERFACE

ITestThreadPoolExecutorService Interface is created.

package com.otv.srv;
 
import java.util.concurrent.ThreadPoolExecutor;
 
import com.otv.handler.TestRejectedExecutionHandler;
 
/**
 * @author onlinetechvision.com
 * @since 17 Oct 2011
 * @version 1.0.0
 * 
 */
public interface ITestThreadPoolExecutorService {
 
    public ThreadPoolExecutor createNewThreadPool();
 
    public int getCorePoolSize();
 
    public void setCorePoolSize(int corePoolSize);
 
    public int getMaxPoolSize();
 
    public void setMaxPoolSize(int maximumPoolSize);
 
    public long getKeepAliveTime();
 
    public void setKeepAliveTime(long keepAliveTime);
 
    public int getQueueCapacity();
 
    public void setQueueCapacity(int queueCapacity);
 
    public TestRejectedExecutionHandler getTestRejectedExecutionHandler();
 
    public void setTestRejectedExecutionHandler(TestRejectedExecutionHandler testRejectedExecutionHandler);
 
}

STEP 6 : CREATE TestThreadPoolExecutorService CLASS

TestThreadPoolExecutorService Class is created by implementing ITestThreadPoolExecutorService Interface. This class creates a new Thread Pool.

package com.otv.srv;
 
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
 
import com.otv.handler.TestRejectedExecutionHandler;
 
/**
 * @author onlinetechvision.com
 * @since 17 Oct 2011
 * @version 1.0.0
 *
 */
public class TestThreadPoolExecutorService implements ITestThreadPoolExecutorService {
 
    private int  corePoolSize;
    private int  maxPoolSize; 
    private long keepAliveTime;
    private int  queueCapacity;
    TestRejectedExecutionHandler testRejectedExecutionHandler;
     
    public ThreadPoolExecutor createNewThreadPool() {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(getCorePoolSize(), 
                                                                getMaxPoolSize(), 
                                                                getKeepAliveTime(), 
                                                                TimeUnit.SECONDS, 
                                                                new ArrayBlockingQueue<Runnable>(getQueueCapacity()), 
                                                                getTestRejectedExecutionHandler());
        return executor;
    }
     
    public int getCorePoolSize() {
        return corePoolSize;
    }
     
    public void setCorePoolSize(int corePoolSize) {
        this.corePoolSize = corePoolSize;
    }
     
    public int getMaxPoolSize() {
        return maxPoolSize;
    }
     
    public void setMaxPoolSize(int maxPoolSize) {
        this.maxPoolSize = maxPoolSize;
    }
     
    public long getKeepAliveTime() {
        return keepAliveTime;
    }
     
    public void setKeepAliveTime(long keepAliveTime) {
        this.keepAliveTime = keepAliveTime;
    }
     
    public int getQueueCapacity() {
        return queueCapacity;
    }
 
    public void setQueueCapacity(int queueCapacity) {
        this.queueCapacity = queueCapacity;
    }
     
    public TestRejectedExecutionHandler getTestRejectedExecutionHandler() {
        return testRejectedExecutionHandler;
    }
 
    public void setTestRejectedExecutionHandler(TestRejectedExecutionHandler testRejectedExecutionHandler) {
        this.testRejectedExecutionHandler = testRejectedExecutionHandler;
    }
}

STEP 7 : CREATE IThreadPoolMonitorService INTERFACE

IThreadPoolMonitorService Interface is created.

package com.otv.monitor.srv;
 
import java.util.concurrent.ThreadPoolExecutor;
 
public interface IThreadPoolMonitorService extends Runnable {
 
    public void monitorThreadPool();
     
    public ThreadPoolExecutor getExecutor();
     
    public void setExecutor(ThreadPoolExecutor executor);
     
}

STEP 8 : CREATE ThreadPoolMonitorService CLASS

ThreadPoolMonitorService Class is created by implementing IThreadPoolMonitorService Interface. This class monitors created thread pool.

package com.otv.monitor.srv;
 
import java.util.concurrent.ThreadPoolExecutor;
import org.apache.log4j.Logger;
 
/**
 * @author onlinetechvision.com
 * @since 17 Oct 2011
 * @version 1.0.0
 *
 */
public class ThreadPoolMonitorService implements IThreadPoolMonitorService {
     
    private static Logger log = Logger.getLogger(ThreadPoolMonitorService.class);
    ThreadPoolExecutor executor;
    private long monitoringPeriod; 
     
    public void run() {
        try {
            while (true){
                monitorThreadPool();
                Thread.sleep(monitoringPeriod*1000);
            } 
        } catch (Exception e) {
            log.error(e.getMessage());
        }
    }
     
    public void monitorThreadPool() {
        StringBuffer strBuff = new StringBuffer();
        strBuff.append("CurrentPoolSize : ").append(executor.getPoolSize());
        strBuff.append(" - CorePoolSize : ").append(executor.getCorePoolSize());
        strBuff.append(" - MaximumPoolSize : ").append(executor.getMaximumPoolSize());
        strBuff.append(" - ActiveTaskCount : ").append(executor.getActiveCount());
        strBuff.append(" - CompletedTaskCount : ").append(executor.getCompletedTaskCount());
        strBuff.append(" - TotalTaskCount : ").append(executor.getTaskCount());
        strBuff.append(" - isTerminated : ").append(executor.isTerminated());
 
        log.debug(strBuff.toString());
    }
     
    public ThreadPoolExecutor getExecutor() {
        return executor;
    }
 
    public void setExecutor(ThreadPoolExecutor executor) {
        this.executor = executor;
    }   
     
    public long getMonitoringPeriod() {
        return monitoringPeriod;
    }
 
    public void setMonitoringPeriod(long monitoringPeriod) {
        this.monitoringPeriod = monitoringPeriod;
    }
}

STEP 9 : CREATE Starter CLASS

Starter Class is created.

package com.otv.start;
 
import java.util.concurrent.ThreadPoolExecutor;
 
import org.apache.log4j.Logger;
 
import com.otv.handler.TestRejectedExecutionHandler;
import com.otv.monitor.srv.IThreadPoolMonitorService;
import com.otv.monitor.srv.ThreadPoolMonitorService;
import com.otv.srv.ITestThreadPoolExecutorService;
import com.otv.srv.TestThreadPoolExecutorService;
import com.otv.task.TestTask;
 
/**
 * @author onlinetechvision.com
 * @since 17 Oct 2011
 * @version 1.0.0
 *
 */
public class Starter {
     
    private static Logger log = Logger.getLogger(TestRejectedExecutionHandler.class);
     
    IThreadPoolMonitorService threadPoolMonitorService;
    ITestThreadPoolExecutorService testThreadPoolExecutorService;
     
    public void start() {
         
        // A new thread pool is created...
        ThreadPoolExecutor executor = testThreadPoolExecutorService.createNewThreadPool();
        executor.allowCoreThreadTimeOut(true);
         
        // Created executor is set to ThreadPoolMonitorService...
        threadPoolMonitorService.setExecutor(executor);
         
        // ThreadPoolMonitorService is started...
        Thread monitor = new Thread(threadPoolMonitorService);
        monitor.start();
 
        // New tasks are executed...
        for(int i=1;i<10;i++) {
            executor.execute(new TestTask("Task"+i));
        }
         
        try {
            Thread.sleep(40000);
        } catch (Exception e)   {
            log.error(e.getMessage());
        }
         
        for(int i=10;i<19;i++) {
            executor.execute(new TestTask("Task"+i));
        }
         
        // executor is shutdown...
        executor.shutdown();
    }   
 
    public IThreadPoolMonitorService getThreadPoolMonitorService() {
        return threadPoolMonitorService;
    }
 
    public void setThreadPoolMonitorService(IThreadPoolMonitorService threadPoolMonitorService) {
        this.threadPoolMonitorService = threadPoolMonitorService;
    }
 
    public ITestThreadPoolExecutorService getTestThreadPoolExecutorService() {
        return testThreadPoolExecutorService;
    }
 
    public void setTestThreadPoolExecutorService(ITestThreadPoolExecutorService testThreadPoolExecutorService) {
        this.testThreadPoolExecutorService = testThreadPoolExecutorService;
    }
}

STEP 10 : CREATE Application CLASS

Application Class is created. This class runs the application.

package com.otv.start;
 
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
 
/**
 * @author onlinetechvision.com
 * @since 17 Oct 2011
 * @version 1.0.0
 *
 */
public class Application {
     
    public static void main(String[] args) {
       ApplicationContext context = new ClassPathXmlApplicationContext("applicationContext.xml");
       Starter starter = (Starter) context.getBean("Starter");
       starter.start();
    }
     
}

STEP 11 : CREATE applicationContext.xml

applicationContext.xml is created.

<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://www.springframework.org/schema/beans
 
http://www.springframework.org/schema/beans/spring-beans-3.0.xsd">
 
    <!-- Beans Declaration -->
    <bean id="TestTask" class="com.otv.task.TestTask"></bean>
    <bean id="ThreadPoolMonitorService" class="com.otv.monitor.srv.ThreadPoolMonitorService">
        <property name="monitoringPeriod"  value="5" />
    </bean>
    <bean id="TestRejectedExecutionHandler" class="com.otv.handler.TestRejectedExecutionHandler"></bean>
    <bean id="TestThreadPoolExecutorService" class="com.otv.srv.TestThreadPoolExecutorService">
        <property name="corePoolSize"  value="1" />
        <property name="maxPoolSize"   value="3" />
        <property name="keepAliveTime" value="10" />
        <property name="queueCapacity" value="3" />
        <property name="testRejectedExecutionHandler" ref="TestRejectedExecutionHandler" />
    </bean>
    <bean id="Starter" class="com.otv.start.Starter">
        <property name="threadPoolMonitorService" ref="ThreadPoolMonitorService" />
        <property name="testThreadPoolExecutorService" ref="TestThreadPoolExecutorService" />
    </bean>
</beans>

STEP 12 : ALTERNATIVE METHOD TO CREATE THREAD POOL

ThreadPoolTaskExecutor Class provided by Spring can also be used to create Thread Pool.

<bean id="threadPoolTaskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
  <property name="corePoolSize"  value="1" />
  <property name="maxPoolSize"   value="3" />
  <property name="queueCapacity" value="3" />
</bean>
 
<bean id="testTaskExecutor" class="TestTaskExecutor">
  <constructor-arg ref="threadPoolTaskExecutor" />
</bean>

STEP 13 : BUILD PROJECT

After OTV_Spring_ThreadPool Project is build, OTV_Spring_ThreadPool-0.0.1-SNAPSHOT.jar will be created.

STEP 14 : RUN PROJECT

After created OTV_Spring_ThreadPool-0.0.1-SNAPSHOT.jar file is run, below output logs will be shown :

18.10.2011 20:08:48 DEBUG (TestRejectedExecutionHandler.java:19) - Task7 : has been rejected
18.10.2011 20:08:48 DEBUG (TestRejectedExecutionHandler.java:19) - Task8 : has been rejected
18.10.2011 20:08:48 DEBUG (TestRejectedExecutionHandler.java:19) - Task9 : has been rejected
18.10.2011 20:08:48 DEBUG (TestTask.java:25) - Task1 : is started.
18.10.2011 20:08:48 DEBUG (TestTask.java:25) - Task6 : is started.
18.10.2011 20:08:48 DEBUG (ThreadPoolMonitorService.java:39) - CurrentPoolSize : 3 - CorePoolSize : 1 - MaximumPoolSize : 3 - ActiveTaskCount : 2 - CompletedTaskCount : 0 - TotalTaskCount : 5 - isTerminated : false
18.10.2011 20:08:48 DEBUG (TestTask.java:25) - Task5 : is started.
18.10.2011 20:08:53 DEBUG (ThreadPoolMonitorService.java:39) - CurrentPoolSize : 3 - CorePoolSize : 1 - MaximumPoolSize : 3 - ActiveTaskCount : 3 - CompletedTaskCount : 0 - TotalTaskCount : 6 - isTerminated : false
18.10.2011 20:08:58 DEBUG (TestTask.java:27) - Task6 : is completed.
18.10.2011 20:08:58 DEBUG (TestTask.java:27) - Task1 : is completed.
18.10.2011 20:08:58 DEBUG (TestTask.java:25) - Task3 : is started.
18.10.2011 20:08:58 DEBUG (TestTask.java:25) - Task2 : is started.
18.10.2011 20:08:58 DEBUG (ThreadPoolMonitorService.java:39) - CurrentPoolSize : 3 - CorePoolSize : 1 - MaximumPoolSize : 3 - ActiveTaskCount : 3 - CompletedTaskCount : 2 - TotalTaskCount : 6 - isTerminated : false
18.10.2011 20:08:58 DEBUG (TestTask.java:27) - Task5 : is completed.
18.10.2011 20:08:58 DEBUG (TestTask.java:25) - Task4 : is started.
18.10.2011 20:09:03 DEBUG (ThreadPoolMonitorService.java:39) - CurrentPoolSize : 3 - CorePoolSize : 1 - MaximumPoolSize : 3 - ActiveTaskCount : 3 - CompletedTaskCount : 3 - TotalTaskCount : 6 - isTerminated : false
18.10.2011 20:09:08 DEBUG (TestTask.java:27) - Task2 : is completed.
18.10.2011 20:09:08 DEBUG (TestTask.java:27) - Task3 : is completed.
18.10.2011 20:09:08 DEBUG (TestTask.java:27) - Task4 : is completed.
18.10.2011 20:09:08 DEBUG (ThreadPoolMonitorService.java:39) - CurrentPoolSize : 3 - CorePoolSize : 1 - MaximumPoolSize : 3 - ActiveTaskCount : 0 - CompletedTaskCount : 6 - TotalTaskCount : 6 - isTerminated : false
18.10.2011 20:09:13 DEBUG (ThreadPoolMonitorService.java:39) - CurrentPoolSize : 3 - CorePoolSize : 1 - MaximumPoolSize : 3 - ActiveTaskCount : 0 - CompletedTaskCount : 6 - TotalTaskCount : 6 - isTerminated : false
18.10.2011 20:09:18 DEBUG (ThreadPoolMonitorService.java:39) - CurrentPoolSize : 0 - CorePoolSize : 1 - MaximumPoolSize : 3 - ActiveTaskCount : 0 - CompletedTaskCount : 6 - TotalTaskCount : 6 - isTerminated : false
18.10.2011 20:09:23 DEBUG (ThreadPoolMonitorService.java:39) - CurrentPoolSize : 0 - CorePoolSize : 1 - MaximumPoolSize : 3 - ActiveTaskCount : 0 - CompletedTaskCount : 6 - TotalTaskCount : 6 - isTerminated : false
18.10.2011 20:09:28 DEBUG (TestTask.java:25) - Task10 : is started.
18.10.2011 20:09:28 DEBUG (TestRejectedExecutionHandler.java:19) - Task16 : has been rejected
18.10.2011 20:09:28 DEBUG (TestRejectedExecutionHandler.java:19) - Task17 : has been rejected
18.10.2011 20:09:28 DEBUG (TestRejectedExecutionHandler.java:19) - Task18 : has been rejected
18.10.2011 20:09:28 DEBUG (TestTask.java:25) - Task14 : is started.
18.10.2011 20:09:28 DEBUG (TestTask.java:25) - Task15 : is started.
18.10.2011 20:09:28 DEBUG (ThreadPoolMonitorService.java:39) - CurrentPoolSize : 3 - CorePoolSize : 1 - MaximumPoolSize : 3 - ActiveTaskCount : 3 - CompletedTaskCount : 6 - TotalTaskCount : 12 - isTerminated : false
18.10.2011 20:09:33 DEBUG (ThreadPoolMonitorService.java:39) - CurrentPoolSize : 3 - CorePoolSize : 1 - MaximumPoolSize : 3 - ActiveTaskCount : 3 - CompletedTaskCount : 6 - TotalTaskCount : 12 - isTerminated : false
18.10.2011 20:09:38 DEBUG (TestTask.java:27) - Task10 : is completed.
18.10.2011 20:09:38 DEBUG (TestTask.java:25) - Task11 : is started.
18.10.2011 20:09:38 DEBUG (TestTask.java:27) - Task14 : is completed.
18.10.2011 20:09:38 DEBUG (TestTask.java:27) - Task15 : is completed.
18.10.2011 20:09:38 DEBUG (TestTask.java:25) - Task12 : is started.
18.10.2011 20:09:38 DEBUG (TestTask.java:25) - Task13 : is started.
18.10.2011 20:09:38 DEBUG (ThreadPoolMonitorService.java:39) - CurrentPoolSize : 3 - CorePoolSize : 1 - MaximumPoolSize : 3 - ActiveTaskCount : 3 - CompletedTaskCount : 9 - TotalTaskCount : 12 - isTerminated : false
18.10.2011 20:09:43 DEBUG (ThreadPoolMonitorService.java:39) - CurrentPoolSize : 3 - CorePoolSize : 1 - MaximumPoolSize : 3 - ActiveTaskCount : 3 - CompletedTaskCount : 9 - TotalTaskCount : 12 - isTerminated : false
18.10.2011 20:09:48 DEBUG (TestTask.java:27) - Task11 : is completed.
18.10.2011 20:09:48 DEBUG (TestTask.java:27) - Task13 : is completed.
18.10.2011 20:09:48 DEBUG (TestTask.java:27) - Task12 : is completed.
18.10.2011 20:09:48 DEBUG (ThreadPoolMonitorService.java:39) - CurrentPoolSize : 0 - CorePoolSize : 1 - MaximumPoolSize : 3 - ActiveTaskCount : 0 - CompletedTaskCount : 12 - TotalTaskCount : 12 - isTerminated : true
18.10.2011 20:09:53 DEBUG (ThreadPoolMonitorService.java:39) - CurrentPoolSize : 0 - CorePoolSize : 1 - MaximumPoolSize : 3 - ActiveTaskCount : 0 - CompletedTaskCount : 12 - TotalTaskCount : 12 - isTerminated : true

STEP 15 : DOWNLOAD

OTV_Spring_ThreadPool

Thread pool Spring Framework Monitor (synchronization) Interface (computing)

Published at DZone with permission of Eren Avsarogullari, DZone MVB. See the original article here.

Opinions expressed by DZone contributors are their own.

Related

Trending

How to Develop and Monitor Thread Pool Services Using Spring

Related

Partner Resources