网站建设长春,学校网站建设实训总结,重庆网站建设 狐灵科技,济南 规划 网站目录 线程池
自定义线程池
步骤1#xff1a;自定义拒绝策略接口
步骤2#xff1a;自定义任务队列
步骤3#xff1a;自定义线程池 步 骤 4#xff1a;测 试
ThreadPoolExecutor 线程池状态 构造方法
工作方式
newFixedThreadPool
newCachedThreadPool
newSingleTh…目录 线程池
自定义线程池
步骤1自定义拒绝策略接口
步骤2自定义任务队列
步骤3自定义线程池 步 骤 4测 试
ThreadPoolExecutor 线程池状态 构造方法
工作方式
newFixedThreadPool
newCachedThreadPool
newSingleThreadExecutor 提交任务 1.执行任务
2.提交任务task用返回值Future获得任务执行结果
3.提交tasks中所有的任务
4.提交 tasks 中所有任务带超时时间
5.提交 tasks 中所有任务哪个任务先成功执行完毕返回此任务执行结果其它任务取消
6.提交 tasks 中所有任务哪个任务先成功执行完毕返回此任务执行结果其它任务取消带超时时间
关闭线程池
shutdown
shutdownNow
任务调度线程池
Timer
ScheduledExecutorService
Tomcat 线程池 Fork/Join
概念
使用 线程池
自定义线程池 步骤1自定义拒绝策略接口
FunctionalInterface // 拒绝策略
interface RejectPolicyT {void reject(BlockingQueueT queue, T task);
}
步骤2自定义任务队列
class BlockingQueueT{//1.任务队列private DequeT queue new ArrayDeque();//2.锁private ReentrantLock lock new ReentrantLock();//生产者条件变量private Condition fullWaitSet lock.newCondition();//4.消费者条件变量private Condition emptyWaitSet lock.newCondition();//5.容量private int capcity;public BlockingQueue(int capcity){this.capcity capcity;}//超时的阻塞添加public T poll(long timeout, TimeUnit unit){lock.lock();try{//将时间统一转化为纳秒long nanos unit.toNanos(timeout);while(queue.isEmpty()){try {if(nanos0){return null;}//awaitNanos()方法返回的时间是一个剩余时间把返回的时间重新赋值给它防止虚假唤醒从头再等nanos emptyWaitSet.awaitNanos(nanos);} catch (InterruptedException e) {throw new RuntimeException(e);}}T t queue.removeFirst();fullWaitSet.signal();return t;}finally {lock.unlock();}}//阻塞获取public T take(){//获取任务的时候需要保证线程安全需要加一把锁lock.lock();try{while(queue.isEmpty()){try {emptyWaitSet.await();} catch (InterruptedException e) {throw new RuntimeException(e);}}T t queue.removeFirst();fullWaitSet.signal();return t;}finally {lock.unlock();}}//阻塞添加public void put (T element){lock.lock();try{while(queue.size()capcity){try {fullWaitSet.await();} catch (InterruptedException e) {throw new RuntimeException(e);}}queue.addLast(element);emptyWaitSet.signal();}finally {lock.unlock();}}//获取当前队列的大小public int size(){lock.lock();try {return queue.size();}finally {lock.unlock();}}
}
步骤3自定义线程池
class ThreadPool{//阻塞队列private BlockingQueueRunnable taskQueue;//线程集合private HashSetWorker workers new HashSet();//核心线程数private int coreSize;//获取任务的超时时间private long timeout;//时间单位private TimeUnit timeUnit;//执行任务public void excute(Runnable task){//当任务数没有超过coreSize时直接交给worker执行//当任务数超过coreSize时候加入任务队列暂存synchronized (workers){if(workers.size()coreSize){Worker worker new Worker(task);log.debug(新增 worker{},任务对象{},worker,task);workers.add(worker);worker.start();}else {log.debug(加入任务队列{},task);taskQueue.put(task);}}}//构造方法public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit,int queueCapcity) {this.coreSize coreSize;this.timeout timeout;this.timeUnit timeUnit;this.taskQueue new BlockingQueue(queueCapcity);}class Worker extends Thread{private Runnable task;public Worker(Runnable task){this.tasktask;}Overridepublic void run(){//执行任务//当task不为空执行任务//当task执行完毕再接着从任务队列获取任务并执行
// while (task !null ||(task taskQueue.take()) !null){while (task !null ||(task taskQueue.poll(timeout,timeUnit)) !null){try {log.debug(正在执行...{},task);task.run();}catch (Exception e){e.printStackTrace();}finally {task null;}}synchronized (workers){log.debug(worker 被移除{},this);workers.remove(this);}}}
} 步 骤 4测 试
public class Test {public static void main(String[] args) {ThreadPool threadPool new ThreadPool(2, 1000, TimeUnit.MILLISECONDS, 10);for (int i 0; i 5; i) {int j i;threadPool.excute(()-{log.debug({},j);});}}
}
以上是一个基本的不借助不借助jdk提供线程池框架的简单例子可以自行多练习几遍有助于理解jdk提供线程框架的底层。
ThreadPoolExecutor 线程池状态
ThreadPoolExecutor使用int的高3位来表示线程池状态低29位表示线程数量
状态名高3位接收新任务处理阻塞队列任务说明RUNNING111YYSHUTDOWN000NY不会接收新任务但是会处理阻塞队列剩余任务STOP001NN会中断正在执行的任务并抛弃阻塞队列任务TIDYING010--任务执行完毕活动线程为0即将进入终结TERMINATED011--终结状态 从数字上比较TERMINATED TIDYING STOP SHUTDOWN RUNNING
这些信息存储在一个原子变量ctl中目的是将线程状态与线程个数合二为一这样就可以用一次cas原子操作进行赋值
// c 为旧值 ctlOf 返回结果为新值
ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))));// rs 为高 3 位代表线程池状态 wc 为低 29 位代表线程个数ctl 是合并它们
private static int ctlOf(int rs, int wc) { return rs | wc; }构造方法
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueueRunnable workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler)
corePoolSize核心线程数目最多保留的线程数maximumPoolSize最大线程数目keepAliveTime生存时间-针对救急线程unit 时间单位-针对救急线程workQueue阻塞队列threadFactory线程工厂-可以为线程创建的时候起名字handler拒绝策略
工作方式 1.线程池中刚开始没有线程当一个任务提交给线程池后线程池会创建一个新线程来执行任务
2.当线程数达到corePoolSize的时候并没有线程空闲再来新任务的时候会加入任务队列workQueue队列排队直到有空闲的线程
3.如果队列选择了有界队列那么任务超过了队列大小时会创建maximumPoolSize-corePoolSize的数量的线程来救急。
4.如果线程达到了maximumPoolSize的时候仍有新任务这时候会执行拒绝策略。拒绝策略jdk提供了4种实现其他著名框架也提供了实现 AbortPolicy 让调用者抛出 RejectedExecutionException 异常这是默认策略CallerRunsPolicy 让调用者运行任务DiscardPolicy 放弃本次任务DiscardOldestPolicy 放弃队列中最早的任务本任务取而代之Dubbo 的实现在抛出 RejectedExecutionException 异常之前会记录日志并 dump 线程栈信息方 便定位问题Netty 的实现是创建一个新线程来执行任务ActiveMQ 的实现带超时等待60s尝试放入队列类似我们之前自定义的拒绝策略PinPoint 的实现它使用了一个拒绝策略链会逐一尝试策略链中每种拒绝策略
5.当高峰过去后超过corePoolSize的救急线程如果一段时间没有任务做需要结束节省资源这个时间由keepAliveTime和unit控制
根据这个构造方法JDK Executors 类中提供了众多工厂方法来创建各种用途的线程池
newFixedThreadPool
public static ExecutorService newFixedThreadPool(int nThreads) {return new ThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueueRunnable());
}
特点
核心线程数最大线程数没有救急线程因此也无需超时时间阻塞队列是无界的可以放任意数据的任务适用于任务量已知相对耗时的任务
newCachedThreadPool
public static ExecutorService newCachedThreadPool() {return new ThreadPoolExecutor( 0, Integer.MAX_VALUE,60L, TimeUnit.SECONDS,new SynchronousQueueRunnable());
}
特点
核心线程数是0最大线程数是Integer.MAX_VALUE,救急线程的生存时间为60s意味着全部都是救急线程60s后可以回收救急线程可以无限创建队列采用了SynchronousQueue实现特点是它没有容量没有线程来取是放不进去的一手交钱一手交货整个线程池表现为线程数会根据任务量不断增长没有上限当任务执行完毕空闲 1分钟后释放线 程。 适合任务数比较密集但每个任务执行时间较短的情况
示例
public class Test2 {public static void main(String[] args) {SynchronousQueueInteger integers new SynchronousQueue();new Thread(()-{try {log.debug(putting{},1);integers.put(1);log.debug(putted{},1);log.debug(putting{},2);integers.put(2);log.debug(putted{},2);} catch (InterruptedException e) {throw new RuntimeException(e);}}).start();}
}
07:37:56.496 [Thread-0] DEBUG com.example.com.yunlong.test3.Test2 - putting1
从上面示例结果可以看出来integers队列把1放进去之后如果没有来取的会直接阻塞到放入元素行不再执行下去。
newSingleThreadExecutor
public static ExecutorService newSingleThreadExecutor() {return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueueRunnable()));
}
使用场景
希望多个任务排队执行。线程数固定为1任务数多余1时会放入无界队列排队。任务执行完毕这唯一的线程也不会被释放
区别
自己创建一个单线程串行执行任务如果任务执行失败而终止那么没有任何补救措施而线程池还会新建一个线程保证池的正常工作
Executors.newSingleThreadExecutor()线程个数始终为1不能修改
FinalizableDelegatedExecutorService 应用的是装饰器模式只对外暴露了 ExecutorService 接口因 此不能调用 ThreadPoolExecutor 中特有的方法
Executors.newFixedThreadPool(1) 初始时为1以后还可以修改
对外暴露的是 ThreadPoolExecutor 对象可以强转后调用 setCorePoolSize 等方法进行修改 提交任务 1.执行任务
void execute(Runnable command);
2.提交任务task用返回值Future获得任务执行结果
T FutureT submit(CallableT task);
示例
public class Test4 {public static void main(String[] args) throws ExecutionException, InterruptedException {ExecutorService executorService Executors.newFixedThreadPool(2);FutureString f executorService.submit(() - {log.debug(running);Thread.sleep(1000);return 1;});//用Future对象调用get()方法拿到结果如果拿不到结果就在这阻塞住log.debug({},f.get());}
} 总结Runnable接口中的唯一抽象方法run()方法没有返回值Callable接口中的唯一抽象方法call()方法有返回值。
FunctionalInterface
public interface Runnable {// 这个run()方法返回值为void没有受检异常的异常声明出现异常只能内部捕获public abstract void run();
}FunctionalInterface
public interface CallableV {// 这个call()方法有返回值且声明了受检异常可以直接抛出Exception异常V call() throws Exception;
}
3.提交tasks中所有的任务
T ListFutureT invokeAll(Collection? extends CallableT tasks)throws InterruptedException;
4.提交 tasks 中所有任务带超时时间
T ListFutureT invokeAll(Collection? extends CallableT tasks,long timeout, TimeUnit unit)throws InterruptedException;
5.提交 tasks 中所有任务哪个任务先成功执行完毕返回此任务执行结果其它任务取消
T T invokeAny(Collection? extends CallableT tasks)throws InterruptedException, ExecutionException;
6.提交 tasks 中所有任务哪个任务先成功执行完毕返回此任务执行结果其它任务取消带超时时间
T T invokeAny(Collection? extends CallableT tasks,long timeout, TimeUnit unit)throws InterruptedException, ExecutionException, TimeoutException;
关闭线程池
shutdown
/*
线程池状态变为 SHUTDOWN- 不会接收新任务- 但已提交任务会执行完- 此方法不会阻塞调用线程的执行
*/
void shutdown();public void shutdown() {final ReentrantLock mainLock this.mainLock;mainLock.lock();try {checkShutdownAccess();// 修改线程池状态advanceRunState(SHUTDOWN);// 仅会打断空闲线程interruptIdleWorkers();onShutdown(); // 扩展点 ScheduledThreadPoolExecutor} finally {mainLock.unlock();}// 尝试终结(没有运行的线程可以立刻终结如果还有运行的线程也不会等)tryTerminate();
}
shutdownNow
/*
线程池状态变为 STOP- 不会接收新任务- 会将队列中的任务返回- 并用 interrupt 的方式中断正在执行的任务
*/
ListRunnable shutdownNow();public ListRunnable shutdownNow() {ListRunnable tasks;final ReentrantLock mainLock this.mainLock;mainLock.lock();try {checkShutdownAccess();// 修改线程池状态advanceRunState(STOP);// 打断所有线程interruptWorkers();// 获取队列中剩余任务tasks drainQueue();} finally {mainLock.unlock();}// 尝试终结tryTerminate();return tasks;
}任务调度线程池
Timer
在任务调度线程池功能加入之前可以使用java.util.Timer来实现定时功能Timer的优点在于简单易用但由于所有任务都是由同一线程来调度因此所有任务都是串行执行的同一时间只能有一个任务在执行前一个任务的延迟或异常都将会影响到之后的任务。
示例
public static void main(String[] args) throws ExecutionException, InterruptedException {Timer timer new Timer();TimerTask task1 new TimerTask() {Overridepublic void run() {log.debug(task1);try {Thread.sleep(2000);} catch (InterruptedException e) {throw new RuntimeException(e);}}};TimerTask task2 new TimerTask() {Overridepublic void run() {log.debug(task2);}};//可以看到task1延迟2秒此时线程会等待task1执行完才能执行后边的任务timer.schedule(task1,1000);timer.schedule(task2,1000);}08:04:04.329 [Timer-0] DEBUG com.example.com.yunlong.test3.Test4 - task1
08:04:06.366 [Timer-0] DEBUG com.example.com.yunlong.test3.Test4 - task2
ScheduledExecutorService
使用ScheduledExecutorService改写
public class Test4 {public static void main(String[] args) throws ExecutionException, InterruptedException {ScheduledExecutorService scheduledExecutorService Executors.newScheduledThreadPool(2);scheduledExecutorService.schedule(()-{log.debug(task1);try {Thread.sleep(1000);} catch (InterruptedException e) {throw new RuntimeException(e);}},1,TimeUnit.SECONDS);scheduledExecutorService.schedule(()-{log.debug(task2);},1,TimeUnit.SECONDS);}
}
08:11:38.793 [pool-1-thread-1] DEBUG com.example.com.yunlong.test3.Test4 - task1
08:11:38.793 [pool-1-thread-2] DEBUG com.example.com.yunlong.test3.Test4 - task2
可以看到前一个任务的执行延迟并不会影响到另外的线程。
Tomcat 线程池
为什么jdk提供了线程池tomcat还要自定义线程池是因为jdk提供的线程池是cpu类型的(cpu计算类型的任务处理较快处理完了可以去queue再取task)而tomcat 处理的请求大多数io相关的如果核心线程满了就入队那io请求就会被阻塞所以tomcat是线程个数到最大线程数之后才会进队列这个和jdk的有点区别 LimitLatch 用来限流可以控制最大连接个数类似 J.U.C 中的 Semaphore 后面再讲Acceptor 只负责【接收新的 socket 连接】Poller 只负责监听 socket channel 是否有【可读的 I/O 事件】 一旦可读封装一个任务对象socketProcessor提交给 Executor 线程池处理Executor 线程池中的工作线程最终负责【处理请求】
Tomcat 线程池扩展了 ThreadPoolExecutor行为稍有不同
如果总线程数达到 maximumPoolSize
这时不会立刻抛 RejectedExecutionException 异常 而是再次尝试将任务放入队列如果还失败才抛出 RejectedExecutionException 异常
public void execute(Runnable command,long timeout,TimeUnit unit){submittedCount.incrementAndGet();try{super.execute(command);}catch(RejectedExecutionException rx){if(super.getQueue()instanceof TaskQueue){final TaskQueue queue(TaskQueue)super.getQueue();try{if(!queue.force(command,timeout,unit)){submittedCount.decrementAndGet();throw new RejectedExecutionException(Queue capacity is full.);}}catch(InterruptedException x){submittedCount.decrementAndGet();Thread.interrupted();throw new RejectedExecutionException(x);}}else{submittedCount.decrementAndGet();throw rx;}}} TaskQueue.java
public boolean force(Runnable o, long timeout, TimeUnit unit) throws InterruptedException {if ( parent.isShutdown() ) throw new RejectedExecutionException(Executor not running, cant force a command into the queue);return super.offer(o,timeout,unit); //forces the item onto the queue, to be used if the task
is rejected
} Executor 线程配置
配置项默认值说明threadPriority5线程优先级daemontrue是否守护线程minSpareThreads25核心线程数即 corePoolSizemaxThreads200最大线程数即 maximumPoolSizemaxIdleTime 60000线程生存时间单位是毫秒默认值即 1 分钟maxQueueSizeInteger.MAX_VALUE队列长度prestartminSpareThreadsfalse核心线程是否在服务器启动时启动 Fork/Join
概念
Fork/Join 是 JDK 1.7 加入的新的线程池实现它体现的是一种分治思想适用于能够进行任务拆分的 cpu 密集型 运算
所谓的任务拆分是将一个大任务拆分为算法上相同的小任务直至不能拆分可以直接求解。跟递归相关的一些计 算如归并排序、斐波那契数列、都可以用分治思想进行求解
Fork/Join 在分治的基础上加入了多线程可以把每个任务的分解和合并交给不同的线程来完成进一步提升了运 算效率
Fork/Join 默认会创建与 cpu 核心数大小相同的线程池
使用
提交给 Fork/Join 线程池的任务需要继承 RecursiveTask有返回值或 RecursiveAction没有返回值例如下 面定义了一个对 1~n 之间的整数求和的任务
public static void main(String[] args) {ForkJoinPool pool new ForkJoinPool(4);System.out.println(pool.invoke(new AddTask1(5)));
}class AddTask1 extends RecursiveTaskInteger {int n;public AddTask1(int n) {this.n n;}Overridepublic String toString() {return { n };}Overrideprotected Integer compute() {// 如果 n 已经为 1可以求得结果了if (n 1) {log.debug(join() {}, n);return n;}[ForkJoinPool-1-worker-0] - fork() 2 {1}
[ForkJoinPool-1-worker-1] - fork() 5 {4}
[ForkJoinPool-1-worker-0] - join() 1
[ForkJoinPool-1-worker-0] - join() 2 {1} 3
[ForkJoinPool-1-worker-2] - fork() 4 {3}
[ForkJoinPool-1-worker-3] - fork() 3 {2}
[ForkJoinPool-1-worker-3] - join() 3 {2} 6
[ForkJoinPool-1-worker-2] - join() 4 {3} 10
[ForkJoinPool-1-worker-1] - join() 5 {4} 15
15// 将任务进行拆分(fork)AddTask1 t1 new AddTask1(n - 1);t1.fork();log.debug(fork() {} {}, n, t1);// 合并(join)结果int result n t1.join();log.debug(join() {} {} {}, n, t1, result);return result;}
}
用图来表示
