最近在学习线程池、内存控制等关于提高程序运行性能方面的编程技术,在网上看到有一哥们写得不错,故和大家一起分享。
[分享]Java 线程池的原理与实现
这几天主要是狂看源程序,在弥补了一些以前知识空白的同时,也学会了不少新的知识(比如 NIO),或者称为新技术吧。
线程池就是其中之一,一提到线程,我们会想到以前《操作系统》的生产者与消费者,信号量,同步控制等等。
一提到池,我们会想到数据库连接池,但是线程池又如何呢?
建议 :在阅读本文前,先理一理同步的知识,特别是syncronized同步关键字的用法。
关于我对同步的认识,要缘于大三年的一本书,书名好像是 Java 实战,这本书写得实在太妙了,真正的从理论到实践,从截图分析到.class字节码分析。哇,我想市场上很难买到这么精致的书了。作为一个Java爱好者,我觉得绝对值得一读。
我对此书印象最深之一的就是:equal()方法,由浅入深,经典!
还有就是同步了,其中提到了我的几个编程误区,以前如何使用同步提高性能等等,通过学习,使我对同步的认识进一步加深。
简单介绍
创建线程有两种方式:继承Thread或实现Runnable。Thread实现了Runnable接口,提供了一个空的run()方法,所以不论是继承Thread还是实现Runnable,都要有自己的run()方法。
一个线程创建后就存在,调用start()方法就开始运行(执行run()方法),调用wait进入等待或调用sleep进入休眠期,顺利运行完毕或休眠被中断或运行过程中出现异常而退出。
wait和sleep比较:
sleep方法有:sleep(long millis),sleep(long millis, long nanos),调用sleep方法后,当前线程进入休眠期,暂停执行,但该线程继续拥有监视资源的所有权。到达休眠时间后线程将继续执行,直到完成。若在休眠期另一线程中断该线程,则该线程退出。
wait方法有:wait(),wait(long timeout),wait(long timeout, long nanos),调用wait方法后,该线程放弃监视资源的所有权进入等待状态;
wait():等待有其它的线程调用notify()或notifyAll()进入调度状态,与其它线程共同争夺监视。wait()相当于wait(0),wait(0, 0)。
wait(long timeout):当其它线程调用notify()或notifyAll(),或时间到达timeout亳秒,或有其它某线程中断该线程,则该线程进入调度状态。
wait(long timeout, long nanos):相当于wait(1000000*timeout + nanos),只不过时间单位为纳秒。
线程池:
多线程技术主要解决处理器单元内多个线程执行的问题,它可以显著减少处理器单元的闲置时间,增加处理器单元的吞吐能力。
假设一个服务器完成一项任务所需时间为:T1 创建线程时间,T2 在线程中执行任务的时间,T3 销毁线程时间。
如果:T1 + T3 远大于 T2,则可以采用线程池,以提高服务器性能。
一个线程池包括以下四个基本组成部分:
1、线程池管理器(ThreadPool):用于创建并管理线程池,包括 创建线程池,销毁线程池,添加新任务;
2、工作线程(PoolWorker):线程池中线程,在没有任务时处于等待状态,可以循环的执行任务;
3、任务接口(Task):每个任务必须实现的接口,以供工作线程调度任务的执行,它主要规定了任务的入口,任务执行完后的收尾工作,任务的执行状态等;
4、任务队列(taskQueue):用于存放没有处理的任务。提供一种缓冲机制。
线程池技术正是关注如何缩短或调整T1,T3时间的技术,从而提高服务器程序性能的。它把T1,T3分别安排在服务器程序的启动和结束的时间段或者一些空闲的时间段,这样在服务器程序处理客户请求时,不会有T1,T3的开销了。
线程池不仅调整T1,T3产生的时间段,而且它还显著减少了创建线程的数目,看一个例子:
假设一个服务器一天要处理50000个请求,并且每个请求需要一个单独的线程完成。在线程池中,线程数一般是固定的,所以产生线程总数不会超过线程池中线程的数目,而如果服务器不利用线程池来处理这些请求则线程总数为50000。一般线程池大小是远小于50000。所以利用线程池的服务器程序不会为了创建50000而在处理请求时浪费时间,从而提高效率。
001 |
/** 线程池类,工作线程作为其内部类 **/ |
002 |
003 |
package org.ymcn.util;
|
004 |
005 |
import java.util.Collections;
|
006 |
import java.util.Date;
|
007 |
import java.util.LinkedList;
|
008 |
import java.util.List;
|
009 |
010 |
import org.apache.log4j.Logger;
|
011 |
012 |
/** |
013 |
* 线程池 |
014 |
* 创建线程池,销毁线程池,添加新任务 |
015 |
* |
016 |
* <SPAN class=referer>@author</SPAN> obullxl |
017 |
*/ |
018 |
public final class ThreadPool {
|
019 |
private static Logger logger = Logger.getLogger(ThreadPool. class );
|
020 |
private static Logger taskLogger = Logger.getLogger( "TaskLogger" );
|
021 |
022 |
private static boolean debug = taskLogger.isDebugEnabled();
|
023 |
// private static boolean debug = taskLogger.isInfoEnabled(); |
024 |
/* 单例 */ |
025 |
private static ThreadPool instance = ThreadPool.getInstance(); |
026 |
027 |
public static final int SYSTEM_BUSY_TASK_COUNT = 150; |
028 |
/* 默认池中线程数 */ |
029 |
public static int worker_num = 5; |
030 |
/* 已经处理的任务数 */ |
031 |
private static int taskCounter = 0; |
032 |
033 |
public static boolean systemIsBusy = false; |
034 |
035 |
private static List<Task> taskQueue = Collections |
036 |
.synchronizedList(new LinkedList<Task>()); |
037 |
/* 池中的所有线程 */ |
038 |
public PoolWorker[] workers; |
039 |
040 |
private ThreadPool() { |
041 |
workers = new PoolWorker[5]; |
042 |
for (int i = 0; i < workers.length; i++) { |
043 |
workers[i] = new PoolWorker(i); |
044 |
} |
045 |
} |
046 |
047 |
private ThreadPool(int pool_worker_num) { |
048 |
worker_num = pool_worker_num; |
049 |
workers = new PoolWorker[worker_num]; |
050 |
for (int i = 0; i < workers.length; i++) { |
051 |
workers[i] = new PoolWorker(i); |
052 |
} |
053 |
} |
054 |
055 |
public static synchronized ThreadPool getInstance() { |
056 |
if (instance == null) |
057 |
return new ThreadPool(); |
058 |
return instance; |
059 |
} |
060 |
/** |
061 |
* 增加新的任务 |
062 |
* 每增加一个新任务,都要唤醒任务队列 |
063 |
* @param newTask |
064 |
*/ |
065 |
public void addTask(Task newTask) { |
066 |
synchronized (taskQueue) { |
067 |
newTask.setTaskId(++taskCounter); |
068 |
newTask.setSubmitTime(new Date()); |
069 |
taskQueue.add(newTask); |
070 |
/* 唤醒队列, 开始执行 */ |
071 |
taskQueue.notifyAll(); |
072 |
} |
073 |
logger.info("Submit Task<" + newTask.getTaskId() + ">: " |
074 |
+ newTask.info()); |
075 |
} |
076 |
/** |
077 |
* 批量增加新任务 |
078 |
* @param taskes |
079 |
*/ |
080 |
public void batchAddTask(Task[] taskes) { |
081 |
if (taskes == null || taskes.length == 0) { |
082 |
return; |
083 |
} |
084 |
synchronized (taskQueue) { |
085 |
for (int i = 0; i < taskes.length; i++) { |
086 |
if (taskes[i] == null) { |
087 |
continue; |
088 |
} |
089 |
taskes[i].setTaskId(++taskCounter); |
090 |
taskes[i].setSubmitTime(new Date()); |
091 |
taskQueue.add(taskes[i]); |
092 |
} |
093 |
/* 唤醒队列, 开始执行 */ |
094 |
taskQueue.notifyAll(); |
095 |
} |
096 |
for (int i = 0; i < taskes.length; i++) { |
097 |
if (taskes[i] == null) { |
098 |
continue; |
099 |
} |
100 |
logger.info("Submit Task<" + taskes[i].getTaskId() + ">: " |
101 |
+ taskes[i].info()); |
102 |
} |
103 |
} |
104 |
/** |
105 |
* 线程池信息 |
106 |
* <SPAN class=referer>@return</SPAN> |
107 |
*/ |
108 |
public String getInfo() { |
109 |
StringBuffer sb = new StringBuffer(); |
110 |
sb.append("\nTask Queue Size:" + taskQueue.size()); |
111 |
for (int i = 0; i < workers.length; i++) { |
112 |
sb.append("\nWorker " + i + " is " |
113 |
+ ((workers[i].isWaiting()) ? "Waiting." : "Running.")); |
114 |
} |
115 |
return sb.toString(); |
116 |
} |
117 |
/** |
118 |
* 销毁线程池 |
119 |
*/ |
120 |
public synchronized void destroy() { |
121 |
for (int i = 0; i < worker_num; i++) { |
122 |
workers[i].stopWorker(); |
123 |
workers[i] = null; |
124 |
} |
125 |
taskQueue.clear(); |
126 |
} |
127 |
128 |
/** |
129 |
* 池中工作线程 |
130 |
* |
131 |
* <SPAN class=referer>@author</SPAN> obullxl |
132 |
*/ |
133 |
private class PoolWorker extends Thread { |
134 |
private int index = -1; |
135 |
/* 该工作线程是否有效 */ |
136 |
private boolean isRunning = true; |
137 |
/* 该工作线程是否可以执行新任务 */ |
138 |
private boolean isWaiting = true; |
139 |
140 |
public PoolWorker(int index) { |
141 |
this.index = index; |
142 |
start(); |
143 |
} |
144 |
145 |
public void stopWorker() { |
146 |
this.isRunning = false; |
147 |
} |
148 |
149 |
public boolean isWaiting() { |
150 |
return this.isWaiting; |
151 |
} |
152 |
/** |
153 |
* 循环执行任务 |
154 |
* 这也许是线程池的关键所在 |
155 |
*/ |
156 |
public void run() { |
157 |
while (isRunning) { |
158 |
Task r = null; |
159 |
synchronized (taskQueue) { |
160 |
while (taskQueue.isEmpty()) { |
161 |
try { |
162 |
/* 任务队列为空,则等待有新任务加入从而被唤醒 */ |
163 |
taskQueue.wait(20); |
164 |
} catch (InterruptedException ie) { |
165 |
logger.error(ie); |
166 |
} |
167 |
} |
168 |
/* 取出任务执行 */ |
169 |
r = (Task) taskQueue.remove(0); |
170 |
} |
171 |
if (r != null) { |
172 |
isWaiting = false; |
173 |
try { |
174 |
if (debug) { |
175 |
r.setBeginExceuteTime(new Date()); |
176 |
taskLogger.debug("Worker<" + index |
177 |
+ "> start execute Task<" + r.getTaskId() + ">"); |
178 |
if (r.getBeginExceuteTime().getTime() |
179 |
- r.getSubmitTime().getTime() > 1000) |
180 |
taskLogger.debug("longer waiting time. " |
181 |
+ r.info() + ",<" + index + ">,time:" |
182 |
+ (r.getFinishTime().getTime() - r |
183 |
.getBeginExceuteTime().getTime())); |
184 |
} |
185 |
/* 该任务是否需要立即执行 */ |
186 |
if (r.needExecuteImmediate()) { |
187 |
new Thread(r).start(); |
188 |
} else { |
189 |
r.run(); |
190 |
} |
191 |
if (debug) { |
192 |
r.setFinishTime(new Date()); |
193 |
taskLogger.debug("Worker<" + index |
194 |
+ "> finish task<" + r.getTaskId() + ">"); |
195 |
if (r.getFinishTime().getTime() |
196 |
- r.getBeginExceuteTime().getTime() > 1000) |
197 |
taskLogger.debug("longer execution time. " |
198 |
+ r.info() + ",<" + index + ">,time:" |
199 |
+ (r.getFinishTime().getTime() - r |
200 |
.getBeginExceuteTime().getTime())); |
201 |
} |
202 |
} catch (Exception e) { |
203 |
e.printStackTrace(); |
204 |
logger.error(e); |
205 |
} |
206 |
isWaiting = true; |
207 |
r = null; |
208 |
} |
209 |
} |
210 |
} |
211 |
} |
212 |
} |
213 |
214 |
/** 任务接口类 **/ |
215 |
216 |
package org.ymcn.util; |
217 |
218 |
import java.util.Date; |
219 |
220 |
/** |
221 |
* 所有任务接口 |
222 |
* 其他任务必须继承访类 |
223 |
* |
224 |
* <SPAN class=referer>@author</SPAN> obullxl |
225 |
*/ |
226 |
public abstract class Task implements Runnable { |
227 |
// private static Logger logger = Logger.getLogger(Task.class); |
228 |
/* 产生时间 */ |
229 |
private Date generateTime = null; |
230 |
/* 提交执行时间 */ |
231 |
private Date submitTime = null; |
232 |
/* 开始执行时间 */ |
233 |
private Date beginExceuteTime = null; |
234 |
/* 执行完成时间 */ |
235 |
private Date finishTime = null; |
236 |
237 |
private long taskId; |
238 |
239 |
public Task() { |
240 |
this.generateTime = new Date(); |
241 |
} |
242 |
243 |
/** |
244 |
* 任务执行入口 |
245 |
*/ |
246 |
public void run() { |
247 |
/** |
248 |
* 相关执行代码 |
249 |
* |
250 |
* beginTransaction(); |
251 |
* |
252 |
* 执行过程中可能产生新的任务 subtask = taskCore(); |
253 |
* |
254 |
* commitTransaction(); |
255 |
* |
256 |
* 增加新产生的任务 ThreadPool.getInstance().batchAddTask(taskCore()); |
257 |
*/ |
258 |
} |
259 |
260 |
/** |
261 |
* 所有任务的核心 所以特别的业务逻辑执行之处 |
262 |
* |
263 |
* @throws Exception |
264 |
*/ |
265 |
public abstract Task[] taskCore() throws Exception; |
266 |
267 |
/** |
268 |
* 是否用到数据库 |
269 |
* |
270 |
* <SPAN class=referer>@return</SPAN> |
271 |
*/ |
272 |
protected abstract boolean useDb(); |
273 |
274 |
/** |
275 |
* 是否需要立即执行 |
276 |
* |
277 |
* <SPAN class=referer>@return</SPAN> |
278 |
*/ |
279 |
protected abstract boolean needExecuteImmediate(); |
280 |
281 |
/** |
282 |
* 任务信息 |
283 |
* |
284 |
* <SPAN class=referer>@return</SPAN> String |
285 |
*/ |
286 |
public abstract String info();
|
287 |
288 |
public Date getGenerateTime() {
|
289 |
return generateTime;
|
290 |
} |
291 |
292 |
public Date getBeginExceuteTime() {
|
293 |
return beginExceuteTime;
|
294 |
} |
295 |
296 |
public void setBeginExceuteTime(Date beginExceuteTime) {
|
297 |
this .beginExceuteTime = beginExceuteTime;
|
298 |
} |
299 |
300 |
public Date getFinishTime() {
|
301 |
return finishTime;
|
302 |
} |
303 |
304 |
public void setFinishTime(Date finishTime) {
|
305 |
this .finishTime = finishTime;
|
306 |
} |
307 |
308 |
public Date getSubmitTime() {
|
309 |
return submitTime;
|
310 |
} |
311 |
312 |
public void setSubmitTime(Date submitTime) {
|
313 |
this .submitTime = submitTime;
|
314 |
} |
315 |
316 |
public long getTaskId() {
|
317 |
return taskId;
|
318 |
} |
319 |
320 |
public void setTaskId( long taskId) {
|
321 |
this .taskId = taskId;
|
322 |
} |
323 |
324 |
} |
相关推荐
Java线程池的原理与实现 Java线程池的原理与实现
JAVA线程池的原理与实现.pdf
介绍了java线程池的基本实现原理并由详细的实例
主要介绍了Java线程池FutureTask实现原理详解,小编觉得还是挺不错的,具有一定借鉴价值,需要的朋友可以参考下
Java语言游戏项目实战资源包 内容概览: 这次分享为你带来了丰富的Java语言游戏项目实战资源,让你在实践中深入...持续学习与探索:Java语言和游戏开发技术都在不断更新,建议你在实践中持续学习新的技术和工具,不
主要为大家详细介绍了Java concurrency线程池之线程池原理,具有一定的参考价值,感兴趣的小伙伴们可以参考一下
2.线程池的运作原理 3.调度线程池的运作原理 4.调度线程池怎么实现FixRate,FixDelay 5.怎么取消的
}关于FutureTask这个类的实现,我在前面的JAVA LOCK代码浅析有讲过其实现原理,主要的思想就是关注任务完成与未完成的状态,任务提交线程get()结
2.Vector与ArrayList一样,也是通过数组实现的,不同的是它支持线程的同步,即某一时刻只有一个线程能够写Vector,避免多线程同时写而引起的不一致性,但实现同步需要很高的花费,因此,访问它比访问ArrayList慢 ...
主要介绍了深入理解Java编程线程池的实现原理,涉及ThreadPoolExecutor类,线程池实现原理及示例等相关内容,具有一定参考价值,需要的朋友可以了解下。
主要介绍了java基于线程池和反射机制实现定时任务的方法,以完整实例形式较为详细的分析了Java定时任务的功能原理与实现技巧,具有一定参考借鉴价值,需要的朋友可以参考下
线程之间通信之join应用与实现原理剖析.mp4 ThreadLocal 使用及实现原理.mp4 并发工具类CountDownLatch详解.mp4 并发工具类CyclicBarrier 详解.mp4 并发工具类Semaphore详解.mp4 并发工具类Exchanger详解.mp4 ...
本文档对Java中的线程和线程池进行了简单介绍。首先,阐述了为什么需要线程、...然后,介绍了为什么需要线程池,JDK自带的线程池实现方式ThreadPoolExecutor的使用及其原理,最后强调ThreadPoolExecutor应用的注意点。
第35节线程之间通信之join应用与实现原理剖析00:10:17分钟 | 第36节ThreadLocal 使用及实现原理00:17:41分钟 | 第37节并发工具类CountDownLatch详解00:22:04分钟 | 第38节并发工具类CyclicBarrier 详解00:11:52...
本节主要关注在配置和调整线程池时用的高级选项,讲述了任务执行框架的过程中需 ... 这就是线程池的实现原理。循环方法中不断获取 Runnable 是用 Queue 实现的,在获 取下一个 Runnable 之前可以是阻塞的。
里面是自己收藏的几篇关于线程的文档资料,大家可以看看哈
3、ThreadLocal 的底层实现与使用 4、ReentrantLock底层实现和如何使用 5、Condition源码分析 6、ReentrantReadWriteLock底层实现原理 7、并发工具类CountDownLatch 、CyclicBarrier和Semaphore底层实现原理 8、...
1. 降低创建线程和销毁线程的性能开销 2. 提高响应速度,当有新任务需要执行是不需要等待线程创建就可以立马执行 3. 合理的设置线程池大小可以避免因为线程数超
在前面的文章中,我们...我们来详细讲解一下Java的线程池,首先我们从核心的ThreadPoolExecutor类中的方法讲起,然后再讲述它的实现原理,接着给出了它的使用示例,后讨论了一下如何合理配置线程池的大小。 以下是