网站建设a云世家网络,小程序开店流程,外链代发公司,注册公司最好用老年人线程池
问题背景
只是单纯使用 new Thread(runnable).start(); 的方式创建线程, 将会导致严重的程序性能问题: 1.线程创建, 销毁需要消耗很大的系统资源; 2.虚拟机创建线程的数量是有限的; 2.线程调度切换也将使程序性能下降; 针对这些问题, 对线程数量进行管理, 有效地重复利…线程池
问题背景
只是单纯使用 new Thread(runnable).start(); 的方式创建线程, 将会导致严重的程序性能问题: 1.线程创建, 销毁需要消耗很大的系统资源; 2.虚拟机创建线程的数量是有限的; 2.线程调度切换也将使程序性能下降; 针对这些问题, 对线程数量进行管理, 有效地重复利用线程, 将会很好地提高程序性能.
线程池原理
使用队列创建一定数量的线程, 当有任务的时候, 使用队列中线程执行任务(如果任务过多, 就将其放入任务队列, 进入等待执行状态), 任务执行完就自动回收线程队列中的线程(任务过少或者任务数量小于线程数量, 超出的线程将会销毁, 做到线程队列具有伸缩性);
根据上面描述, 我们自己的线程池将具有一下特点: 1.内部使用队列来管理线程, 管理提交的任务. 2.控制线程数量, 做到线程队列具有良好的伸缩性. 3.当任务数过多, 或者任务队列已经饱和, 将使用任务拒绝策略, 告诉对应的任务提交者. 4.使用线程工厂定制线程队列中, 每个线程的名字, 状态, 是否为守护线程等等.
线程池类图结构 任务队列, 队列使用limit限制提交任务的大小, 实现RunnableQueue接口(RunnableQueue接口负责: 1.接收用户提交的任务; 2.获取任务队列中的任务; 3.查看任务队列大小), LinkedRunnableQueue实现RunnableQueue中的方法, 并且针对用户提交不同的任务以及线程池种类(ThreadPool)的不同, 决定是否执行拒绝策略(拒绝策略具有多个, 拒绝方式取决于用户自定义, 在线程池内部具有默认的拒绝策略实现); 实现Runnable接口, 在run方法中获取RunnableQueue中的任务, 然后执行RunnQueue中的任务, InternalTask中的run方法是一个while循环循环结束条件取决于是否关闭该线程(关闭线程据需要设置flag变量, 当flage为false, 线程run方法结束, 自动结束生命), 而不是当前用户提交的任务是否执行完!!!; InternalTask主要是对RunnableQueue的一种封装; stop方法主要是设置线程flag(flag主要判断当前线程是否关闭) 线程池原型: 1.实现Runnable接口(此处注明, BasicThreadPool是继承Thread, 但是Thread内容太多了不能很好地在UML图中显示, 所以我就把他删除了, 只留下了实现Runnable接口), 因为线程池自身执行也需要一个线程, 所以继承Thread, 这样可以在BasicThreadPool的构造方法中执行start(), run方法中执行创建线程的操作(线程池内部执行任务的线程); 创建线程取决于线程池设置的最大线程数, 核心线程数, 初始化线程数, 用户提交的任务数; 2.实现ThreadPool接口(该接口主要用于定义线程池的基本操作, 比如执行任务, 获取线程池的一些基本属性) ; 3.内部具有2个内部类(ThreadTask负责对InternalTask进行封装, DefaultThreadFactory主要定义默认的线程创建方式), 不同的线程池中拥有不同的默认创建方式, 因此将线程创建方式设置为内部类; 4.在BasicThreadPool中使用newThread方法创建线程(这些线程用于执行ThreadTask中的任务); 5.线程池原型中具有2个队列, 第一个是刚才上面提的RunnQueue(负责执行的任务), 第二个是ThreadQueue(负责存储创建的每一个线程, 使用ArrayQueue实现, 这样很好地维护管理了线程, 做到资源重用) 6.removeThread方法: 删除多余的线程, 当用户提交的任务数量小于线程池中创建的线程数量, 那么就删除一定数量的线程, 这样才不会浪费线程资源. 7.在构造方法中设置基本属性, 以及当前线程池的拒绝策略.
每个接口, 类的详细定义
ThreadPool(interface 定义线程池基本操作)
package com.concurrent.customthreadpool;/*** 线程池接口* author regotto*/
public interface ThreadPool {/*** 执行提交的Runnable任务* param runnable*/void execute(Runnable runnable);/*** 关闭线程池*/void shutdown();/*** 获得线程池初始化大小* return initSize*/int getInitSize();/*** 获得线程池最大线程数* return maxSize*/int getMaxSize();/*** 获取线程池核心线程数* return coreSize*/int getCoreSize();/*** 获取线程池中用于缓存任务队列的大小* return queueSize*/int getQueueSize();/*** 获取线程池中国活跃的线程数量* return activeCount*/int getActiveCount();/*** 查看线程池是否shutdown* return boolan*/boolean isShutdown();}
RunnableQueue(interface 任务队列)
package com.concurrent.customthreadpool;/*** 存放提交的Runnable, 使用BlockedQueue, 设置limit* author regotto*/
public interface RunnableQueue {/*** 缓存提交到线程池中的任务* param runnable*/void offer(Runnable runnable);/*** 从缓存中获取Runnable任务* 如果没有任务, 调用者线程挂起, 在某些特定的时候抛出中断异常* throws InterruptedException* return runnable*/Runnable take() throws InterruptedException;/*** 缓冲区大小* return size*/int size();}LinkedRunnableQueue(class 对RunnableQueue的封装, 用户提交任务, 线程执行任务, 此过程使用生产者-消费者模式实现)
package com.concurrent.customthreadpool;import java.util.LinkedList;/*** 线程池的内部线程队列, 缓冲区* author regotto*/
public class LinkedRunnableQueue implements RunnableQueue{/*** limit: 限制当前runnableList中还能存放多少内容* denyPolicy: 拒绝策略* runnableList: 存放runnable的缓冲区* threadPool: 线程池*/private final int limit;private final RunnableDenyPolicy denyPolicy;private final LinkedListRunnable runnableList new LinkedList();private final ThreadPool threadPool;public LinkedRunnableQueue(int limit, RunnableDenyPolicy denyPolicy, ThreadPool threadPool){this.limit limit;this.denyPolicy denyPolicy;this.threadPool threadPool;}Overridepublic void offer(Runnable runnable) {synchronized (runnableList) {if (runnableList.size() limit) {//用户提交的任务大于限制条件, 执行对应的拒绝策略System.out.println(runnableList.size() limit execute deny policy);denyPolicy.reject(runnable, threadPool);} else {//添加任务到任务队列尾部, 有任务存在, 唤醒刚才wait的线程runnableList.addLast(runnable);runnableList.notifyAll();}}}Overridepublic Runnable take() throws InterruptedException{synchronized (runnableList) {while (runnableList.isEmpty()) {try {//从RunnableQueue中取出任务, 如果任务为空, 使当前线程waitrunnableList.wait();} catch (InterruptedException e) {throw e;}}//移除任务缓冲区的第一个return runnableList.removeFirst();}}Overridepublic int size() {synchronized (runnableList) {return runnableList.size();}}
}
InternalTask(class 对RunnableQueue中任务的执行)
package com.concurrent.customthreadpool;/*** 用于线程池内部, 获取runnableQueue中的runnable* author regotto*/
public class InternalTask implements Runnable {private final RunnableQueue runnableQueue;private volatile boolean running true;public InternalTask(RunnableQueue runnableQueue){this.runnableQueue runnableQueue;}Overridepublic void run() {//如果线程没有关闭, 就让该线程死循环, 处理每一个提交的任务while (running !Thread.currentThread().isInterrupted()){try {//处于中断时候的线程不做处理//获取RunnableQueue中任务, 然后执行Runnable take runnableQueue.take();System.out.println(runnableQueue.take(): take.toString());take.run();} catch (InterruptedException e) {running false;break;}}}/*** 停止当前任务, 设置其running为false, 在shutdown中处理*/public void stop(){this.running false;}}
RunnableDenyPolicy(interface 任务拒绝策略)
package com.concurrent.customthreadpool;/*** 当任务数提交超过缓冲区limit, 执行对应的任务拒绝策略* author regotto*/
FunctionalInterface
public interface RunnableDenyPolicy {/*** 对提交到threadPool的runnable是否执行reject* param runnable* param threadPool*/void reject(Runnable runnable, ThreadPool threadPool);/*** 该策略使用空方法直接丢弃任务*/class DiscardDenyPolicy implements RunnableDenyPolicy {Overridepublic void reject(Runnable runnable, ThreadPool threadPool) {System.out.println(runnable 不做处理);}}/*** 该策略抛出一个RunnableDenyException*/class AbortDenyPolicy implements RunnableDenyPolicy {Overridepublic void reject(Runnable runnable, ThreadPool threadPool) {throw new RunnableDenyException(The runnable will be abort);}}/***该策略Runnable给提交者所在的线程中运行, 不加入到线程中*/class RunnerDenyPolicy implements RunnableDenyPolicy{Overridepublic void reject(Runnable runnable, ThreadPool threadPool) {if (threadPool.isShutdown()) {runnable.run();}}}}
RunnableDenyException(class 处理RunnableDenyPolicy抛出的运行时异常)
package com.concurrent.customthreadpool;public class RunnableDenyException extends RuntimeException {public RunnableDenyException(String message) {super(message);}
}
ThreadFactory(interface 定义创建线程的接口)
package com.concurrent.customthreadpool;/*** 创建线程接口, 定制线程属于哪一个group, 是否为守护线程, 优先级, 名字等* author regotto*/
public interface ThreadFactory {/*** 创建定制化线程* param runnable* return thread*/Thread createThread(Runnable runnable);}BasicThreadPool(class 线程池的实现)
package com.concurrent.customthreadpool;import java.util.ArrayDeque;
import java.util.Queue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;/*** 默认的自定义的线程池, 内部使用Queue进行维护* author regotto*/
public class BasicThreadPool extends Thread implements ThreadPool{/**initSize: 初始化线程数* maxSize: 线程池最大线程数* coreSize: 线程核心数* activeCount: 当前活跃线程数* threadFactory: 线程工厂, 配置线程创建需要的参数* runnableQueue: 任务队列* isShutdown: 是否关闭线程池* threadQueue: 工作线程队列* DEFAULT_THREAD_FACTORY: 默认的线程工厂* keepAliveTime: 线程存活时间*/private final int initSize;private final int maxSize;private final int coreSize;private int activeCount;private final ThreadFactory threadFactory;private final RunnableQueue runnableQueue;private volatile boolean isShutdown false;private QueueThreadTask threadQueue new ArrayDeque();private final static RunnableDenyPolicy DEFAULT_DENY_POLICY new RunnableDenyPolicy.DiscardDenyPolicy();private final static ThreadFactory DEFAULT_THREAD_FACTORY new DefaultThreadFactory();private final static long DEFAULT_KEEP_ALIVE_TIME 10;private final long keepAliveTime;private final TimeUnit timeUnit;public BasicThreadPool(int initSize, int maxSize, int coreSize, int queueSize) {this(initSize, maxSize, coreSize, DEFAULT_THREAD_FACTORY, queueSize, DEFAULT_DENY_POLICY,DEFAULT_KEEP_ALIVE_TIME, TimeUnit.SECONDS);}public BasicThreadPool(int initSize, int maxSize, int coreSize, ThreadFactory threadFactory, int queueSize,RunnableDenyPolicy denyPolicy, long keepAliveTime, TimeUnit timeUnit) {this.initSize initSize;this.maxSize maxSize;this.coreSize coreSize;this.threadFactory threadFactory;this.runnableQueue new LinkedRunnableQueue(queueSize, denyPolicy, this);this.keepAliveTime keepAliveTime;this.timeUnit timeUnit;this.init();}/*** 初始化线程池, 创建initThread*/private void init() {start();for (int i 0; i initSize; i) {newThread();}}/*** 创建线程添加到线程队列, 然后用该线程执行ThreadTask任务(层层封装, 封装用户提交的任务)*/private void newThread() {InternalTask internalTask new InternalTask(runnableQueue);//使用自定义的线程工厂创建线程Thread thread this.threadFactory.createThread(internalTask);ThreadTask threadTask new ThreadTask(thread, internalTask);System.out.println(threadTask.thread.getName() 被添加);//添加到线程队列threadQueue.offer(threadTask);this.activeCount;//被添加后的线程执行startthread.start();}Overridepublic void execute(Runnable runnable) {if (this.isShutdown) {throw new IllegalStateException(The thread pool id destroy);}//将用户提交的任务放到runnableQueue中, 等待线程队列中线程执行this.runnableQueue.offer(runnable);}private void removeThread() {//ArrayDeque的remove就是removeFirstThreadTask threadTask threadQueue.remove();//设置当前线程flag, 在InternalTask中跳出循环自动结束线程生命threadTask.internalTask.stop();this.activeCount--;}Overridepublic void run() {while (!isShutdown !isInterrupted()) {try {timeUnit.sleep(keepAliveTime);} catch (InterruptedException e) {isShutdown true;break;}synchronized (this) {if (isShutdown) {break;}//当前队列中有任务还没有处理, 且activeCount coreSizeif (runnableQueue.size() 0 activeCount coreSize) {//此处i曾写做i0,导致多创建了一个线程,在没有任务的时候该线程一直保持wait//因为关闭pool,该线程没有add到threadQueue,导致Interrupt失败,最终导致线程一直运行中for (int i initSize; i coreSize; i) {newThread();}//防止后面的if判断创建线程数超过coreSize, 在coreSize还没有满的时候, 只执行当前的ifcontinue;}//当上面if中创建的线程数不足的时候, 就扩大线程池线程数, 直到maxSizeif (runnableQueue.size() 0 activeCount maxSize) {for (int i coreSize; i maxSize; i) {newThread();}}//当没有任务, 但是activeCount线程数超出coreSize大小, 回收超出coreSize的线程if (runnableQueue.size() 0 activeCount coreSize) {for (int i coreSize; i activeCount; i) {removeThread();}}}}}Overridepublic void shutdown() {synchronized (this) {if (!isShutdown) {isShutdown true;System.out.println(threadQueue size: threadQueue.size());threadQueue.forEach(threadTask - {//调用internalTask中stop, 设置当前线程运行标志为falsethreadTask.internalTask.stop();//设置线程中断状态threadTask.thread.interrupt();System.out.println(threadTask.thread.getName());});System.out.println(threadQueue中线程已经关闭);//当前线程池自己也要关闭this.interrupt();}}}Overridepublic int getInitSize() {if (isShutdown) {throw new IllegalStateException(The thread pool is destroy);}return this.initSize;}Overridepublic int getMaxSize() {if (isShutdown) {throw new IllegalStateException(The thread pool is destroy);}return this.maxSize;}Overridepublic int getCoreSize() {if (isShutdown) {throw new IllegalStateException(The thread pool is destroy);}return this.coreSize;}Overridepublic int getQueueSize() {if (isShutdown) {throw new IllegalStateException(The thread pool is destroy);}return this.runnableQueue.size();}Overridepublic int getActiveCount() {if (isShutdown) {throw new IllegalStateException(The thread pool is destroy);}return this.activeCount;}Overridepublic boolean isShutdown() {return this.isShutdown;}/*** 内部类, 定义自己默认的线程工厂*/private static class DefaultThreadFactory implements ThreadFactory {private static final AtomicInteger GROUP_COUNTER new AtomicInteger(1);//设置线程组private static final ThreadGroup GROUP new ThreadGroup(MyThreadPool- GROUP_COUNTER.getAndDecrement());private static final AtomicInteger COUNTER new AtomicInteger(0);Overridepublic Thread createThread(Runnable runnable) {//创建定制化的线程return new Thread(GROUP, runnable, thread-pool- COUNTER.getAndDecrement());}}/*** 封装InternalTask, 与每次创建的线程绑定在一起*/private class ThreadTask {Thread thread;InternalTask internalTask;ThreadTask(Thread thread, InternalTask internalTask){this.thread thread;this.internalTask internalTask;}}
}
ThreadPoolTest(class 测试类)
package com.concurrent.customthreadpool;import java.util.concurrent.TimeUnit;/*** 用于测试自定的线程池* author regotto*/
public class ThreadPoolTest {public static void main(String[] args) {//初始化线程数:2, 最大线程数:6, 核心线程数:4, 任务队列大小:1000final BasicThreadPool threadPool new BasicThreadPool(2, 6, 4, 1000);//创建20个任务提交进行执行for (int i 0; i 20; i) {threadPool.execute(() - {try {TimeUnit.SECONDS.sleep(3);System.out.println(Thread.currentThread().getName() is running and done.);} catch (InterruptedException e) {e.printStackTrace();}});}//此处用于测试线程池运行时基本信息状态
// for (int j 0; j 1000; j) {
// System.out.println(getActiveCount: threadPool.getActiveCount());
// System.out.println(getQueueSize: threadPool.getQueueSize());
// System.out.println(getCoreSize: threadPool.getCoreSize());
// System.out.println(getMaxSize: threadPool.getMaxSize());
// try {
// TimeUnit.SECONDS.sleep(3);
// } catch (InterruptedException e) {
// e.printStackTrace();
// }
// }try {TimeUnit.SECONDS.sleep(25);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(shutdown);//测试线程池shutdown功能threadPool.shutdown();}
}
总结
不要让BasicThreadPool继承Thread, 这样会导致直接调用BasicThreadPool中start方法, 导致线程池崩溃, 应该将Thread设置成组合, 隐藏start();修改拒绝策略, 做到可以让用户自我实现拒绝方式, 此处拒绝策略可以使用策略模式;使用工厂模式定义ThreadPool;对每个参数进行合法验证;在自己写的过程中, 出现参数传递出错的问题, 导致线程一直wait, 如果遇到这种问题, 使用jconsole工具来查看线程状态, 定位出错的位置, 这样能很快解决遇到的问题.
上面有错, 还请指出, 如果认为我写的还不错, 还请点个赞, 多多支持一下, O(∩_∩)O~~
