网站管理 官网,微信公众平台开发者是什么,wordpress文章半透明,网站界面分析预期目标
1.实现一个相对完备的线程池 2.自定义拒绝策略#xff08;下一节#xff09;
线程池的基本参数
1.核心线程数 2.超时时间 3.拒绝策略#xff08;在下一篇中添加#xff09; 4.工作队列 5.任务队列
工作机制
当添加一个任务到线程池中时#xff0c;线程池会…预期目标
1.实现一个相对完备的线程池 2.自定义拒绝策略下一节
线程池的基本参数
1.核心线程数 2.超时时间 3.拒绝策略在下一篇中添加 4.工作队列 5.任务队列
工作机制
当添加一个任务到线程池中时线程池会判断工作线程数量是否小于核心线程数若小于创建工作线程执行任务反之将其添加到任务队列若是当前任务队列已经满了可以执行拒绝策略拒绝策略有很多种例如死等[会阻塞main线程]放弃任务抛出异常等等
工作线程执行过程
工作线程会先将手头上的任务干完然后到工作队列当中取如果工作队列中还有任务取出来继续执行…(周而复始) 但是有可能在一段时间内工作队列中没任务执行这个时候我们可以选择让它死等或者超出指定时间之后自己销毁。
了解这些之后正式开始coding…
1.构建一个阻塞队列
在前面博客中已经实现过了需要锁两个条件变量[生产者消费者]普通队列这三个参数。
Slf4j
class BlockQueueT {//1.任务队列private DequeT tDeque new ArrayDeque();//2.锁private ReentrantLock lock new ReentrantLock();//3.两个条件变量(生产者消费者)private Condition notEmpty;private Condition notFull;private int capacity;public BlockQueue(int capacity) {this.notEmpty lock.newCondition();this.notFull lock.newCondition();this.capacity capacity;}//带超时的阻塞获取public T poll(long timeout, TimeUnit timeUnit) {lock.lock();try {//将timeout转换long nanos timeUnit.toNanos(timeout);while (tDeque.isEmpty()) {try {//返回的是剩余的时间if (nanos 0) return null;nanos notEmpty.awaitNanos(nanos);} catch (InterruptedException e) {log.error(error{},e.getMessage());}}notFull.signal();return tDeque.removeFirst();} finally {lock.unlock();}}//消费者public T take() {lock.lock();try {while (tDeque.isEmpty()) {try {notEmpty.await();} catch (InterruptedException e) {log.error(error{},e.getMessage());}}notFull.signal();return tDeque.removeFirst();//消费对头} finally {lock.unlock();}}//阻塞添加//生产者public void put(T ele) {lock.lock();try {while (tDeque.size() capacity) {try {log.info(等待加入任务队列......);notFull.await();} catch (InterruptedException e) {log.error(error{},e.getMessage());}}log.info(已加入任务队列);tDeque.addLast(ele);notEmpty.signal();} finally {lock.unlock();}}//非阻塞式添加//即使失败也不会阻塞住主线程public boolean offer(T ele, long timeout, TimeUnit timeUnit){lock.lock();try {long nanosTime timeUnit.toNanos(timeout);while (tDeque.size() capacity) {try {if (nanosTime 0) return false;nanosTime notFull.awaitNanos(nanosTime);} catch (InterruptedException e) {log.error(error{},e.getMessage());}}log.info(已加入任务队列);tDeque.addLast(ele);notEmpty.signal();return true;} finally {lock.unlock();}}//获取大小public int size() {lock.lock();try {return tDeque.size();} finally {lock.unlock();}}
}2.写线程池
Slf4j
class ThreadPool {//任务队列private BlockQueueRunnable taskQueue;//线程集合 我们需要对线程做一个包装private HashSetWorker workers new HashSet();//核心线程数量private long coreSize;//超时时间private long timeout;//时间单位private TimeUnit timeUnit;//自定义拒绝策略//private RejectPolicy rejectPolicy;public ThreadPool(int queueCapacity,long coreSize,long timeout,TimeUnit timeUnit){taskQueue new BlockQueue(queueCapacity);this.coreSize coreSize;this.timeout timeout;this.timeUnit timeUnit;}//执行任务public void execute(Runnable task){//当任务数量尚未超过coreSizesynchronized (workers){if (workers.size() coreSize){log.info(创建工作线程{},task);Worker worker new Worker(task);workers.add(worker);worker.start();}else{log.info(加入到任务队列{},task);//有可能会阻塞在这里 进而将主线程阻塞掉taskQueue.put(task);//这里会有很多种策略自定义策略//1.死等//2.带超时等待//3.让调用者放弃任务执行//4.让调用者抛出异常//5.让调用者自己执行任务//策略模式操作抽象成接口实现代码是传过来不会写死}}}class Worker extends Thread{private Runnable task;public Worker(Runnable task){this.task task;}Overridepublic void run() {while (task ! null || (task taskQueue.poll(timeout,timeUnit)) ! null){try {log.info(正在执行...{},task);//执行任务task.run();}catch (Exception e){System.out.println(e.getMessage());}finally {//不要忘记这一步task null;}}synchronized (workers){log.info(worker被移除{},this);workers.remove(this);}}}
}测试:
[main] INFO com.define.ThreadPool - 创建工作线程com.define.TestPool$$Lambda$1/188058798165b3120a
[main] INFO com.define.ThreadPool - 创建工作线程com.define.TestPool$$Lambda$1/18805879814783da3f
[main] INFO com.define.ThreadPool - 加入到任务队列com.define.TestPool$$Lambda$1/188058798149097b5d
[Thread-0] INFO com.define.ThreadPool - 正在执行...com.define.TestPool$$Lambda$1/188058798165b3120a
[main] INFO com.define.BlockQueue - 已加入任务队列
[Thread-1] INFO com.define.ThreadPool - 正在执行...com.define.TestPool$$Lambda$1/18805879814783da3f
[main] INFO com.define.ThreadPool - 加入到任务队列com.define.TestPool$$Lambda$1/18805879816e2c634b
[main] INFO com.define.BlockQueue - 已加入任务队列
[main] INFO com.define.ThreadPool - 加入到任务队列com.define.TestPool$$Lambda$1/188058798137a71e93
[main] INFO com.define.BlockQueue - 已加入任务队列
[main] INFO com.define.ThreadPool - 加入到任务队列com.define.TestPool$$Lambda$1/18805879817e6cbb7a
[main] INFO com.define.BlockQueue - 已加入任务队列
[main] INFO com.define.ThreadPool - 加入到任务队列com.define.TestPool$$Lambda$1/18805879817c3df479
[main] INFO com.define.BlockQueue - 已加入任务队列
[main] INFO com.define.ThreadPool - 加入到任务队列com.define.TestPool$$Lambda$1/18805879817106e68e
[main] INFO com.define.BlockQueue - 已加入任务队列
[main] INFO com.define.ThreadPool - 加入到任务队列com.define.TestPool$$Lambda$1/18805879817eda2dbb
[main] INFO com.define.BlockQueue - 已加入任务队列
[main] INFO com.define.ThreadPool - 加入到任务队列com.define.TestPool$$Lambda$1/18805879816576fe71
[main] INFO com.define.BlockQueue - 已加入任务队列
[main] INFO com.define.ThreadPool - 加入到任务队列com.define.TestPool$$Lambda$1/188058798176fb509a
[main] INFO com.define.BlockQueue - 已加入任务队列
[main] INFO com.define.ThreadPool - 加入到任务队列com.define.TestPool$$Lambda$1/1880587981300ffa5d
[main] INFO com.define.BlockQueue - 已加入任务队列
[main] INFO com.define.ThreadPool - 加入到任务队列com.define.TestPool$$Lambda$1/18805879811f17ae12
[main] INFO com.define.BlockQueue - 等待加入任务队列......测试没什么问题但是能发现如果当前工作线程都是busy,并且任务队列也满了当执行put的时候就会阻塞在这里put阻塞—execute阻塞----main线程阻塞当然阻塞也是一种方式那如果不想让它阻塞比如我添加不进去想让他直接丢弃或者抛出异常应该怎么办那就需要自定义一套拒绝策略下一节继续。
