public interface Executor {
void execute(Runnable command);
}
public interface ExecutorService extends Executor {
void shutdown();
List<Runnable> shutdownNow();
boolean isShutdown();
boolean isTerminated();
boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException;
<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Future<?> submit(Runnable task);
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException;
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException;
<T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException;
<T> T invokeAny(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
}
public class ThreadPoolExecutor extends AbstractExecutorService {
}
public abstract class AbstractExecutorService implements ExecutorService{
}
public class Executors{
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
}
真正的线程池接口是 ExecutorService；
ExecutorService 的默认实现是 ThreadPoolExecutor；
普通类 Executors 里面调用的就是 ThreadPoolExecutor。
Executors 提供四种线程池：
1）newCachedThreadPool 是一个可根据需要创建新线程的线程池，但是在以前构造的线程可用时将重用它们。对于执行很多短期异步任务的程序而言，这些线程池通常可提高程序性能。
调用 execute() 将重用以前构造的线程（如果线程可用）。如果现有线程没有可用的，则创建一个新线程并添加到池中。终止并从缓存中移除那些已有 60 秒钟未被使用的线程。
因此，长时间保持空闲的线程池不会使用任何资源。注意，可以使用 ThreadPoolExecutor 构造方法创建具有类似属性但细节不同（例如超时参数）的线程池。
2）newSingleThreadExecutor 创建是一个单线程池，也就是该线程池只有一个线程在工作，所有的任务是串行执行的，如果这个唯一的线程因为异常结束，
那么会有一个新的线程来替代它，此线程池保证所有任务的执行顺序按照任务的提交顺序执行。
3）newFixedThreadPool 创建固定大小的线程池，每次提交一个任务就创建一个线程，直到线程达到线程池的最大大小，
线程池的大小一旦达到最大值就会保持不变，如果某个线程因为执行异常而结束，那么线程池会补充一个新线程。
4）newScheduledThreadPool 创建一个大小无限的线程池，此线程池支持定时以及周期性执行任务的需求。
通过 ThreadPoolExecutor 的构造函数，撸一撸线程池相关参数的概念：
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
threadFactory, defaultHandler);
}
1）corePoolSize：线程池的核心线程数，一般情况下不管有没有任务都会一直在线程池中一直存活，只有在 ThreadPoolExecutor
中的方法 allowCoreThreadTimeOut(boolean value) 设置为 true 时，闲置的核心线程会存在超时机制，如果在指定时间没有新任务来时，核心线程也会被终止，而这个时间间隔由第3个属性 keepAliveTime 指定。
2）maximumPoolSize：线程池所能容纳的最大线程数，当活动的线程数达到这个值后，后续的新任务将会被阻塞。
3）keepAliveTime：控制线程闲置时的超时时长，超过则终止该线程。一般情况下用于非核心线程，只有在 ThreadPoolExecutor 中的方法 allowCoreThreadTimeOut(boolean value) 设置为 true时，也作用于核心线程。
4）unit：用于指定 keepAliveTime 参数的时间单位，TimeUnit 是个 enum 枚举类型，常用的有：TimeUnit.HOURS(小时)、TimeUnit.MINUTES(分钟)、TimeUnit.SECONDS(秒) 和 TimeUnit.MILLISECONDS(毫秒)等。
5）workQueue：线程池的任务队列，通过线程池的 execute(Runnable command) 方法会将任务 Runnable 存储在队列中。
6）threadFactory：线程工厂，它是一个接口，用来为线程池创建新线程的。
线程池的关闭
ThreadPoolExecutor 提供了两个方法，用于线程池的关闭，分别是 shutdown() 和 shutdownNow()。
shutdown()：不会立即的终止线程池，而是要等所有任务缓存队列中的任务都执行完后才终止，但再也不会接受新的任务。
shutdownNow()：立即终止线程池，并尝试打断正在执行的任务，并且清空任务缓存队列，返回尚未执行的任务。
下面是一个项目中的工具类
package com.sf.module.routecal.routecheck.util;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.sf.module.ompshare.util.SystemUtils;
public class TaskManager {
private static final int DEFAULT_POOL_SIZE = 5;
private static Logger logger = LoggerFactory.getLogger(TaskManager.class);
private static TaskManager taskManager;
// 线程池
private ThreadPoolExecutor threadPool;
// 每个主线程能够同时创建的子线程数
private int concurrentSubTask;
// 任务控制器
private Map<Long, TaskControl> taskControl;
// 任务队列的缺省对象
private Object NULL = new Object();
// 关闭任务
private boolean shutdown;
// 休眠时间
private int sleepTimeWhenNeed = 500;
//休眠5秒
private int sleepTime = 5;
public static TaskManager getInstance() {
if (taskManager == null) {
synchronized (TaskManager.class) {
if (taskManager == null) {
taskManager = new TaskManager(getDefaultTaskCount());
}
}
}
return taskManager;
}
public static TaskManager newInstance(int maxTask) {
if (maxTask < 5) {
maxTask = getDefaultTaskCount();
}
return new TaskManager(maxTask);
}
public static TaskManager newInstanceByTaskNum(int maxTask){
if(taskManager == null){
synchronized (TaskManager.class) {
if (taskManager == null) {
taskManager = new TaskManager(maxTask);
}
}
}
return taskManager;
}
private TaskManager(int maxTask) {
this.taskControl = new ConcurrentHashMap<Long, TaskControl>();
int maxPoolSize = maxTask;
int corePoolSize = maxTask;
this.concurrentSubTask = maxTask;
if (maxTask <= 0) {
corePoolSize = DEFAULT_POOL_SIZE;
maxPoolSize = corePoolSize;
this.concurrentSubTask = corePoolSize / 2;
if (this.concurrentSubTask < 1) {
this.concurrentSubTask = 1;
}
}
this.shutdown = false;
this.threadPool = new ThreadPoolExecutor(corePoolSize, // core pool size
maxPoolSize, // max pool size
10, // alive time: 10 seconds
TimeUnit.SECONDS,//
new TaskBlockingQueue<Runnable>(), //
new TaskPolicy());
}
/**
* 缺省任务数
*
* @return
*/
public static int getDefaultTaskCount() {
return SystemUtils.getProcessorCount();
}
public void shutdown() {
shutdown = true;
threadPool.shutdown();
}
/**
* 获取任务控制器
*
* @return
*/
private TaskControl getTaskControl() {
long threadId = Thread.currentThread().getId();
TaskControl ctrl = taskControl.get(threadId);
if (ctrl == null) {
ctrl = new TaskControl(concurrentSubTask, getCaller());
taskControl.put(threadId, ctrl);
}
return ctrl;
}
private String getCaller() {
StackTraceElement[] stack = (new Throwable()).getStackTrace();
if ((stack != null) && (stack.length > 3)) {
StackTraceElement ste = stack[3];
return ste.getMethodName();
}
return "";
}
/**
* 设置当前线程的任务超时时间
*
* @param timeout
* @param raiseError
*/
public void setTimeout(long timeout, boolean raiseError) {
getTaskControl().setTimeout(System.currentTimeMillis() + timeout, raiseError);
}
/**
* 设置是否显示任务完成信息
*
* @param showInfo
*/
public void setShowInfo(boolean showInfo) {
getTaskControl().setShowInfo(showInfo);
}
/**
* 设置异常类型
*
* @param errorClazz
*/
public void setErrorClass(Class<? extends RuntimeException> errorClazz) {
getTaskControl().setErrorClass(errorClazz);
}
/**
* 在线程池空闲时增加任务
*
* @param task
*/
public void executeTaskWhileNoFull(final Runnable task) {
final TaskControl ctrl = getTaskControl();
if (ctrl.incrementTask()) {
try {
threadPool.execute(new Runnable() {
public void run() {
try {
task.run();
} catch (Exception e) {
ctrl.incrementFailTask(e);
logger.error(String.format("[%s] task run error.", ctrl.getCallerInfo()), e);
} finally {
ctrl.decrementTask();
}
}
});
} catch (Exception e) {
logger.error("threadPool execute error", e);
ctrl.decrementTask();
ctrl.incrementFailTask(e);
}
}
}
/**
* 获得当前线程所发起的任务活动数量
*
* @return
*/
public int getCurrentThreadActiveCount() {
long threadId = Thread.currentThread().getId();
TaskControl ctrl = taskControl.get(threadId);
return (ctrl == null) ? 0 : ctrl.getTaskCount();
}
/**
* 等待直到本线程发起的任务完成
*/
public void waitWhileTaskFinish() {
long threadId = Thread.currentThread().getId();
TaskControl ctrl = taskControl.get(threadId);
if (ctrl != null) {
try {
boolean isTimeout = false;
while ((ctrl.getTaskCount() > 0) && !shutdown && !isTimeout) {
sleep();
isTimeout = ctrl.checkTimeout();
}
ctrl.showInfo();
} finally {
taskControl.remove(threadId);
}
}
}
/**
* 等待直到本线程发起的任务完成
*/
public void waitTaskFinish() {
long threadId = Thread.currentThread().getId();
TaskControl ctrl = taskControl.get(threadId);
if (ctrl != null) {
try {
boolean isTimeout = false;
while ((ctrl.getTaskCount() > 0) && !shutdown && !isTimeout) {
try {
Thread.sleep(sleepTime);
} catch (Exception e) {
logger.error("task waiting error.", e);
}
isTimeout = ctrl.checkTimeout();
}
ctrl.showInfo();
} finally {
taskControl.remove(threadId);
}
}
}
/**
* 等待直到本线程发起的任务完成
*
* @param waitMultiple
* 打印日志需要等待的睡眠次数
* @param waitMessage
* 日志信息
*/
public void waitWhileTaskFinish(int waitMultiple, String waitMessage) {
long threadId = Thread.currentThread().getId();
TaskControl ctrl = taskControl.get(threadId);
if (ctrl != null) {
try {
boolean isTimeout = false;
int count = 0;
while ((ctrl.getTaskCount() > 0) && !shutdown && !isTimeout) {
sleep();
if (++count % waitMultiple == 0) {
logger.info(waitMessage);
count = 0;
}
isTimeout = ctrl.checkTimeout();
}
ctrl.showInfo();
} finally {
taskControl.remove(threadId);
}
}
}
public int getConcurrentSubTask() {
return concurrentSubTask;
}
private void sleep() {
try {
Thread.sleep(sleepTimeWhenNeed);
} catch (Exception e) {
logger.error("task waiting error.", e);
}
}
class TaskBlockingQueue<E> extends LinkedBlockingQueue<E> {
private static final long serialVersionUID = 1L;
public boolean offer(E o) {
if (threadPool.getPoolSize() < threadPool.getMaximumPoolSize()) {
return false;
}
return super.offer(o);
}
}
class TaskPolicy implements RejectedExecutionHandler {
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
executor.getQueue().add(r);
}
}
class TaskControl {
private long startTime;
private AtomicInteger failTask;
private AtomicInteger totalTask;
private BlockingQueue<Object> ctrl;
private String callerInfo;
private Throwable error;
private long timeout;
private boolean raiseError;
private boolean showTimeout;
private boolean showInfo;
private Class<? extends RuntimeException> errorClazz;
public TaskControl(int concurrentSubTask, String callerInfo) {
this.startTime = System.currentTimeMillis();
this.failTask = new AtomicInteger(0);
this.totalTask = new AtomicInteger(0);
this.callerInfo = callerInfo;
this.timeout = 0;
this.raiseError = false;
this.showTimeout = false;
this.showInfo = true;
this.errorClazz = null;
this.ctrl = new ArrayBlockingQueue<Object>(concurrentSubTask);
}
public void incrementFailTask(Throwable error) {
failTask.incrementAndGet();
synchronized (this) {
if ((this.error == null)) {
this.error = error;
}
}
}
public boolean incrementTask() {
boolean isTimeout = false;
try {
isTimeout = checkTimeout();
while (!isTimeout && !shutdown && !ctrl.offer(NULL, 10, TimeUnit.SECONDS)) {
isTimeout = checkTimeout();
}
if (!isTimeout) {
totalTask.incrementAndGet();
}
} catch (Exception e) {
throw new RuntimeException(e);
}
return !isTimeout;
}
public void decrementTask() {
ctrl.poll();
}
public int getTaskCount() {
return ctrl.size();
}
public String getCallerInfo() {
return callerInfo;
}
public void setTimeout(long timeout, boolean raiseError) {
this.timeout = timeout;
this.raiseError = raiseError;
}
public void setShowInfo(boolean showInfo) {
this.showInfo = showInfo;
}
public void setErrorClass(Class<? extends RuntimeException> errorClazz) {
this.errorClazz = errorClazz;
}
private boolean checkTimeout() {
boolean isTimeout = (timeout > 0) && (System.currentTimeMillis() > timeout);
if (isTimeout) {
if (!showTimeout) {
synchronized (this) {
if (!showTimeout) {
showTimeout = true;
double time = (System.currentTimeMillis() - this.startTime) / 1000.0;
String msg = String.format("[%s] task is timeout: %.2f seconds.", callerInfo, time);
if (raiseError) {
throw (error != null) ? new RuntimeException(msg, error) : new RuntimeException(msg);
} else {
logger.warn(msg);
}
}
}
}
}
return isTimeout;
}
public void showInfo() {
if (failTask.get() > 0) {
if (errorClazz != null) {
try {
String msg = String.format("[%s] run task fail count: %d", callerInfo, failTask.get());
throw errorClazz.getConstructor(String.class, Throwable.class).newInstance(msg, error);
} catch (Exception e) {
throw new RuntimeException(e);
}
} else {
String msg = String.format("[%s] run task fail count: %d", callerInfo, failTask.get());
throw new RuntimeException(msg, error);
}
}
if (showInfo) {
double time = (System.currentTimeMillis() - this.startTime) / 1000.0;
logger.info(String.format("[%s] run %d task in %.2f seconds.", callerInfo, totalTask.get(), time));
}
}
}
}
相关推荐
线程池线程池线程池线程池线程池线程池线程池线程池线程池线程池线程池线程池线程池线程池线程池线程池线程池线程池线程池线程池线程池线程池线程池线程池线程池线程池线程池线程池线程池线程池线程池线程池线程池...
线程池是一种多线程处理形式，通过预先创建一定数量的线程并管理它们，以提高系统的效率和响应性。在计算机科学中，特别是在软件开发领域，线程池是操作系统或者编程语言中的一种资源管理技术。它允许程序预先启动一...
corePoolSize：核心池的大小，在创建了线程池后，线程池中的线程数为0，当有任务来之后，就会创建一个线程去执行任务，当线程池中的线程数目达到corePoolSize后，就会把到达的任务放到缓存队列当中; ...
阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池...
一、要实现高效的线程池，可以考虑以下几点 二、实现线程池可以按照以下步骤进行 三、简单的C++线程池代码示例 四、 基于boost编写的源码库 - 线程池 4.1 基于boost编写的源码库地址 4.2 boost线程池的先进先出、...
### 线程池原理及创建（C++实现） #### 一、线程池的重要性 在现代计算环境中，网络服务器面临着处理大量并发请求的挑战，其中包括但不限于Web服务器、电子邮件服务器和数据库服务器。这类服务器通常需要在短时间...
java线程池使用后到底要关闭吗 java线程池是一种高效的并发编程技术，可以帮助开发者更好地管理线程资源，提高系统的性能和可靠性。然而，在使用java线程池时，一个常见的问题是：使用完线程池后到底要不要关闭？...
文章通过实例展示了如何创建一个全局线程池类，该类中封装了线程池对象，并提供了向线程池提交任务、检查任务是否在运行等方法。全局线程池的生命周期与Django主线程的生命周期一致，确保了线程资源的合理释放。 5....
线程池是多线程编程中的一个重要概念，它是一种线程使用模式，通过预先创建一组线程并维护一个线程集合来处理并发任务。在Windows操作系统中，内建的线程池API（Thread Pool API）提供了高效且灵活的线程管理机制，...
线程池是一种在多线程编程中非常重要的概念，它能有效地管理和调度系统中的线程资源，从而提高系统的效率和响应速度。在这个简单的线程池实现中，我们可以通过`pthread_pool.cpp`、`MainFunctionForTest.cpp`、`...
Java8并行流中自定义线程池操作示例 Java8并行流中自定义线程池操作示例主要介绍了Java8并行流中自定义线程池操作，结合实例形式分析了并行流的相关概念、定义及自定义线程池的相关操作技巧。 1. 概览 Java8引入了...
在编程领域，线程池是一种多线程处理形式，处理过程中将任务添加到队列，然后在创建线程后自动启动这些任务。线程池在C++中是提高程序效率和资源管理的重要工具，尤其在处理大量并发操作时。本文将深入探讨VC++中...
Linux 线程池创建 C 实现 线程池是一种常用的并发编程技术，它可以提高应用程序的性能和响应速度。在 Linux 系统中，使用 C 语言创建线程池可以实现高效的并发处理。 什么时候需要创建线程池呢？简单的说，如果一...
DELPHI的线程池（ThreadPool）是一种高效管理并发任务的技术，它允许程序在需要时创建线程，而不是每次需要执行任务时都手动创建。线程池通过预先创建一组线程，然后根据需要分配任务，减少了线程创建和销毁的开销，...
本篇文章将重点探讨两种线程池实现：精易模块线程池和鱼刺模块线程池，并通过源码分析来展示它们的特点和用法。 首先，精易模块（SanYe Module）是由中国程序员SanYe开发的一系列开源模块，其中包含了线程池的实现...
在Linux系统中，线程池是一种高效的进程管理方式，它允许多个任务并行执行，同时限制了系统中并发线程的数量，以优化资源分配和调度。本项目实现了利用线程池进行目录拷贝的功能，这涉及到多个重要的编程概念和技术...
线程池是一种优化资源管理的机制，通过预先创建并维护一组可重用的线程，避免频繁地创建和销毁线程带来的性能开销。在Java、C++等编程语言中，线程池广泛应用于并发处理，提高系统效率，降低系统的资源消耗。本项目...
一、线程池 1、为什么需要使用线程池 1.1 创建/销毁线程伴随着系统开销，过于频繁的创建/销毁线程，会很大程度上影响处理效率。 记创建线程消耗时间T1，执行任务消耗时间T2，销毁线程消耗时间T3，如果T1+T3>T2，那...
线程池管理和多线程上传是并发编程中的一个重要实践，特别是在大数据传输和网络服务中。在Java等编程语言中，线程池通过有效地管理和复用线程资源，避免了频繁创建和销毁线程带来的开销，提升了系统性能。下面将详细...
在C#编程中，线程池（ThreadPool）是一种高效的线程管理机制，它允许开发者创建并管理多个线程，而无需直接操作线程对象。线程池中的线程可以复用，减少了创建和销毁线程的开销。当我们需要执行大量短生命周期的任务...