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后，就会把到达的任务放到缓存队列当中; ...
阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池阻塞线程池...
一、线程池 1、为什么需要使用线程池 1.1 创建/销毁线程伴随着系统开销，过于频繁的创建/销毁线程，会很大程度上影响处理效率。 记创建线程消耗时间T1，执行任务消耗时间T2，销毁线程消耗时间T3，如果T1+T3>T2，那...
VC实现线程池
易语言简易线程池的实现。 ——V雪落有声V原创。转载请保留。前文：。为了能充分理解本篇文章的内容，需要了解的知识如下：。1.事件对象的使用：http://baike.baidu.com/view/751499.htm。2.信号量的使用：...
简单的线程池程序+中文文档 包结构: com.tangkai.threadpool --SimpleThread.java 工作线程 --TestThreadPool.java 程序入口 --ThreadPoolManager.java 线程池管理类
Windows下一个比较完美的线程池实现和示例 本线程池提供了如下功能： 1.能根据任务个数和当前线程的多少在最小/最大线程个数之间自动调整(Vista后的系统有 SetThreadpoolThreadMaximum 等函数有类似功能)； 2.能方便...
C#管理线程池的类 /* How to use Thread Classs * * ============== * public ELMService() { InitializeComponent(); etm.ClalThreadPool("EmailThreads", (uint)ApplicationInfo.EmailParsingThreads); ...
线程池
Spring Boot中内置了多种线程池，为应用程序提供快速响应和高吞吐量的运行环境。线程池在Spring Boot中起着至关重要的作用，它能够有效地管理和复用线程，降低系统的开销。本文将详细介绍线程池的基本参数、每一种...
因为本人是个小白，多线程经常用，但是线程池并没有用过，（一听到线程池，总感觉高大上）。但是近期写彩票软件的时候发现，多线程长期操作会导致内容不断的升高直至报错，遂想起了线程池，完善后发现不是一般的叼...
2.然后根据提示运行java命令执行示例程序，观看线程池的运行结果 目标：Java中多线程技术是一个难点，但是也是一个核心技术。因为Java本身就是一个多线程语言。本人目前在给46班讲授Swing的网络编程--使用Swing来...
Java 实现通用线程池 Java 实现通用线程池是指使用 Java 语言编写一个通用的线程池，线程池通俗的描述就是预先创建若干空闲线程，等到需要用多线程去处理事务的时候去唤醒某些空闲线程执行处理任务，这样就省去了...
线程池技术在并发服务器中的应用 线程池技术是指在服务器程序中，预先创建一组线程，供客户端请求时使用，以减少创建和销毁线程的开销，提高系统的处理能力。这种技术可以最大程度地利用系统的资源，消除系统因频繁...
一、要实现高效的线程池，可以考虑以下几点 二、实现线程池可以按照以下步骤进行 三、简单的C++线程池代码示例 四、 基于boost编写的源码库 - 线程池 4.1 基于boost编写的源码库地址 4.2 boost线程池的先进先出、...
"线程池技术在网络游戏服务器中的应用" 线程池技术是网络游戏服务器中的一种重要技术，它可以提高服务器的性能，减少系统资源的开销。在网络游戏服务器中，线程池技术可以用于处理大量的数据包，以提高服务器的...
1.线程池管理器（ThreadPoolManager）:用于创建并管理线程池 2.工作线程（WorkThread）: 线程池中线程 3.任务接口（Task）:每个任务必须实现的接口，以供工作线程调度任务的执行。 4.任务队列:用于存放没有处理的...
本文将详细介绍线程池的实现原理、线程池的优势、线程池的类型、线程池的创建方式以及线程池的使用注意事项。 一、线程池的实现原理 线程池的实现原理基于池化技术，即减少每次获取资源的消耗，提高资源利用率。...
以下示例显示如何使用线程池。首先创建 ManualResetEvent 对象，此对象使程序能够知道线程池何时运行完所有的工作项。接着，尝试向线程池添加一个线程。如果添加成功，则添加其余的线程（本例中为 4 个）。然后...