并发线程池和阻塞式线程池实现

Executors创建线程池

java的线程池有很多种，最便捷的就是Executors创建线程池，直接指定线程池大小就可以使用了，比如Executors.newFixedThreadPool(int nThreads)，但是便捷不仅隐藏了复杂性，也为我们埋下了潜在的隐患（OOM，线程耗尽）。

当然了，小程序使用这些快捷方法没什么问题，对于服务端需要长期运行的程序，创建线程池应该直接使用ThreadPoolExecutor的构造方法。没错，上述Executors方法创建的线程池就是ThreadPoolExecutor。

ThreadPoolExecutor构造方法

/*** 线程池初始化方法* <p>* corePoolSize 核心线程池大小----10* maximumPoolSize 最大线程池大小----30* keepAliveTime 线程池中超过corePoolSize数目的空闲线程最大存活时间----30+单位TimeUnit* TimeUnit keepAliveTime时间单位----TimeUnit.MINUTES* workQueue 阻塞队列----new ArrayBlockingQueue<Runnable>(10)====10容量的阻塞队列* threadFactory 新建线程工厂----new CustomThreadFactory()====定制的线程工厂* rejectedExecutionHandler 当提交任务数超过maxmumPoolSize+workQueue之和时,* 即当提交第41个任务时(前面线程都没有执行完,此测试方法中用sleep(100)),* 任务会交给RejectedExecutionHandler来处理*/ThreadPoolExecutor pool = new ThreadPoolExecutor(10, 30, 30,TimeUnit.MINUTES, new ArrayBlockingQueue<Runnable>(10), new CustomThreadFactory(), new CustomRejectedExecutionHandler());

可以向线程池提交的任务有两种：Runnable和Callable，二者的区别如下：

1. 方法签名不同，void Runnable.run(), V Callable.call() throws Exception
2. 是否允许有返回值，Callable允许有返回值
3. 是否允许抛出异常，Callable允许抛出异常。

Callable是JDK1.5时加入的接口，作为Runnable的一种补充，允许有返回值，允许抛出异常。

所以，创建好线程后，然后线程池submit就可以了

这里我们定制属于自己的非阻塞线程池

package demo3;import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;/*** 定制属于自己的非阻塞线程池*/
public class CustomThreadPoolExecutor {private ThreadPoolExecutor pool = null;/*** 线程池初始化方法* <p>* corePoolSize 核心线程池大小----10* maximumPoolSize 最大线程池大小----30* keepAliveTime 线程池中超过corePoolSize数目的空闲线程最大存活时间----30+单位TimeUnit* TimeUnit keepAliveTime时间单位----TimeUnit.MINUTES* workQueue 阻塞队列----new ArrayBlockingQueue<Runnable>(10)====10容量的阻塞队列* threadFactory 新建线程工厂----new CustomThreadFactory()====定制的线程工厂* rejectedExecutionHandler 当提交任务数超过maxmumPoolSize+workQueue之和时,* 即当提交第41个任务时(前面线程都没有执行完,此测试方法中用sleep(100)),* 任务会交给RejectedExecutionHandler来处理*/public void init() {pool = new ThreadPoolExecutor(10, 30, 30,TimeUnit.MINUTES, new ArrayBlockingQueue<Runnable>(10), new CustomThreadFactory(), new CustomRejectedExecutionHandler());}public void destory() {if (pool != null) {pool.shutdownNow();}}public ExecutorService getCustomThreadPoolExecutor() {return this.pool;}private class CustomRejectedExecutionHandler implements RejectedExecutionHandler {@Overridepublic void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {//记录异常System.out.println("error...................");}}private class CustomThreadFactory implements ThreadFactory {private AtomicInteger count = new AtomicInteger(0);@Overridepublic Thread newThread(Runnable r) {Thread t = new Thread(r);String threadName = CustomThreadPoolExecutor.class.getSimpleName() + count.addAndGet(1);System.out.println(threadName);t.setName(threadName);return t;}}public static void main(String[] args) {CustomThreadPoolExecutor exec = new CustomThreadPoolExecutor();//1. 初始化exec.init();ExecutorService pool = exec.getCustomThreadPoolExecutor();for (int i = 1; i < 100; i++) {System.out.println("提交第" + i + "个任务");int finalI = i;pool.execute(new Runnable() {@Overridepublic void run() {try {System.out.println(">>>task is running========" + finalI);Thread.sleep(100);} catch (InterruptedException e) {e.printStackTrace();}}});}}

这种是并发式的线程，添加进去的线程会同步执行，如果要实现一个个执行线程，就要定制属于自己的阻塞线程池

阻塞式的线程池核心是一个队列，当然了，普通线程也是一个队列，但是我们将添加的方式有offer转变为out，线程就只能一个一个执行了，

package demo3;import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;/*** @Author:Zach* @Description: 定制属于自己的阻塞线程池* @Date:Created in 15:26 2018/8/14* @Modified By:*/
public class CustomUnblockThreadPoolExecutor {private ThreadPoolExecutor pool = null;/*** 线程池初始化方法* <p>* corePoolSize 核心线程池大小----1* maximumPoolSize 最大线程池大小----3* keepAliveTime 线程池中超过corePoolSize数目的空闲线程最大存活时间----30+单位TimeUnit* TimeUnit keepAliveTime时间单位----TimeUnit.MINUTES* workQueue 阻塞队列----new ArrayBlockingQueue<Runnable>(5)==== 5容量的阻塞队列* threadFactory 新建线程工厂----new CustomThreadFactory()====定制的线程工厂* rejectedExecutionHandler 当提交任务数超过maxmumPoolSize+workQueue之和时,* 即当提交第9个任务时(前面线程都没有执行完,此测试方法中用sleep(100)),* 任务会交给RejectedExecutionHandler来处理*/public void init() {pool = new ThreadPoolExecutor(1, 3, 30,TimeUnit.MINUTES, new ArrayBlockingQueue<Runnable>(5), new CustomThreadFactory(), new CustomRejectedExecutionHandler());}public void destory() {if (pool != null) {pool.shutdownNow();}}public ExecutorService getCustomThreadPoolExecutor() {return this.pool;}private class CustomRejectedExecutionHandler implements RejectedExecutionHandler {@Overridepublic void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {//核心改造点,由blockingqueue的offer改成put阻塞方法try {executor.getQueue().put(r);} catch (InterruptedException e) {e.printStackTrace();}}}private class CustomThreadFactory implements ThreadFactory {private AtomicInteger count = new AtomicInteger(0);@Overridepublic Thread newThread(Runnable r) {Thread t = new Thread(r);String threadName = CustomUnblockThreadPoolExecutor.class.getSimpleName() + count.addAndGet(1);System.out.println(threadName);t.setName(threadName);return t;}}public static void main(String[] args) {CustomUnblockThreadPoolExecutor exec = new CustomUnblockThreadPoolExecutor();//1. 初始化exec.init();ExecutorService pool = exec.getCustomThreadPoolExecutor();for (int i = 1; i < 100; i++) {System.out.println("提交第" + i + "个任务");pool.execute(new Runnable() {@Overridepublic void run() {try {System.out.println(">>>task is running========");TimeUnit.SECONDS.sleep(10);} catch (InterruptedException e) {e.printStackTrace();}}});}}