45、线程池ThreadPoolExecutor
45.1、创建线程池
Java通过Executors提供四个静态方法创建四种线程池，分别为：
newCachedThreadPool创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。
newFixedThreadPool 创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。
newScheduledThreadPool 创建一个定长线程池，支持定时及周期性任务执行。
newSingleThreadExecutor 创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO， LIFO，优先级)执行。
45.2、关键参数解析
corepoolsize：核心池的大小，默认情况下，在创建了线程池之后，线程池中线程数为 0，当有任务来之后，就会创建一个线程去执行任务，当线程池中线程数达到 corepoolsize 后，就把任务放在任务缓存队列中。
Maximumpoolsize：线程池中最多创建多少个线程。
Keeplivetime：线程没有任务执行时，最多保存多久的时间会终止，默认情况下，当线程池中线程数 > corepoolsize时，Keeplivetime才起作用，直到线程数不大于 corepoolsize。
workQueue：阻塞队列，用来存放等待被执行的任务。
threadFactory：线程工厂，用来创建线程。
45.3、线程池的状态
1.当线程池创建后，初始为 running 状态；
2.调用 shutdown 方法后，处 shutdown 状态，此时不再接受新的任务，等待已有的任务执行完毕；
3.调用 shutdownnow 方法后，进入 stop 状态，不再接受新的任务，并且会尝试终止正在执行的任务。
4.当处于 shotdown 或 stop 状态，并且所有工作线程已经销毁，任务缓存队列已清空，线程池被设为 terminated 状态。
45.4、当有任务提交到线程池之后的一些操作：
1.若当前线程池中线程数 < corepoolsize，则每来一个任务就创建一个线程去执行。
2.若当前线程池中线程数 >= corepoolsize，会尝试将任务添加到任务缓存队列中去，若添加成功，则任务会等待空闲线程将其取出执行，若添加失败，则尝试创建线程去执行这个任务。
3.若当前线程池中线程数 >= Maximumpoolsize，则采取拒绝策略。
1）、abortpolicy：丢弃任务，抛出 RejectedExecutionException；
2）、discardpolicy：拒绝执行，不抛异常；
3）、discardoldestpolicy：丢弃任务缓存队列中最老的任务，并且尝试重新提交新的任务；
4）、callerrunspolicy：由调用线程处理该任务，有反馈机制，使任务提交的速度变慢。
46、生产者消费者模式实现
1、阻塞队列BlockingQueue实现生产者消费者模式
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ArrayBlockingQueue;
/**
* @Description: 阻塞队列实现生产者消费者模式
*/
public class ProducerConsumerPattern {
public static void main(String[] args) {
BlockingQueue<String> sharedQueue = new ArrayBlockingQueue<String>(2);
// ArrayBlockingQueue：需要设置队列大小，LinkedBlockingQueue不设置的话默认大小为Integer.MAX_VALUE
// BlockingQueue<String> sharedQueue2 = new LinkedBlockingQueue<String>(2);
Producer producer = new Producer(sharedQueue);
Consumer consumer = new Consumer(sharedQueue);
for(int i = 0; i < 5; i++) {
new Thread(producer, "Producer" + (i + 1)).start();
new Thread(consumer, "Consumer" + (i + 1)).start();
}
}
}
// 生产者
class Producer implements Runnable {
private BlockingQueue<String> sharedQueue;
public Producer(BlockingQueue<String> sharedQueue) {
this.sharedQueue = sharedQueue;
}
@Override
public void run() {
try {
String prod = "产品：" + Thread.currentThread().getName();
// 如果队列是满的话，会阻塞当前线程
sharedQueue.put(prod);
System.out.println("我是生产线程，生产了一个产品：" + prod);
} catch (Exception e) {
System.out.println(e);
}
}
}
// 消费者
class Consumer implements Runnable {
private BlockingQueue<String> sharedQueue;
public Consumer(BlockingQueue<String> sharedQueue) {
this.sharedQueue = sharedQueue;
}
@Override
public void run() {
try {
// 如果队列为空，会阻塞当前线程
String prod = sharedQueue.take();
System.out.println("我是消费者，消费产品: " + prod);
} catch (Exception e) {
System.out.println(e);
}
}
}
2、使用Condition实现
import java.util.PriorityQueue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
* @Description: Condition实现生产者消费者模式
*/
public class ProducerConsumerPattern2 {
private PriorityQueue<String> queue = new PriorityQueue<String>(3);
private Lock lock = new ReentrantLock();
private Condition notFull = lock.newCondition();
private Condition notEmpty = lock.newCondition();
public static void main(String[] args) {
ProducerConsumerPattern2 test = new ProducerConsumerPattern2();
Producer producer = test.new Producer();
Consumer consumer = test.new Consumer();
for(int i = 0; i < 5; i++) {
new Thread(producer, "Producer" + (i + 1)).start();
new Thread(consumer, "Consumer" + (i + 1)).start();
}
}
class Producer implements Runnable {
@Override
public void run() {
lock.lock();
try {
while(queue.size() == 3) {
try {
notFull.await();
} catch (Exception e) {
e.printStackTrace();
}
}
// 每次插入一个元素
String prod = "" + Thread.currentThread().getName();
queue.offer(prod);
System.out.println("我是生产线程，生产了一个产品：" + prod);
// 通知队列不空
notEmpty.signal();
} finally {
lock.unlock();
}
}
}
class Consumer implements Runnable {
@Override
public void run() {
lock.lock();
try {
while(queue.size() == 0) {
try {
notEmpty.await();
} catch (Exception e) {
e.printStackTrace();
}
}
String prod = queue.poll();
System.out.println("我是消费者，消费产品: " + prod);
// 通知队列未满
notFull.signal();
} finally {
lock.unlock();
}
}
}
}
网友评论