为什么使用线程池？
在实际使用中，线程是很占用系统资源的，如果对线程管理不善
很容易导致系统问题。因此，在大多数并发框架中都会使用线程
池来管理线程，使用线程池管理线程主要有如下好处：
1、使用线程池可以重复利用已有的线程继续执行任务，避免线程在创建和销毁时造成的消耗
2、由于没有线程创建和销毁时的消耗，可以提高系统响应速度
3、通过线程可以对线程进行合理的管理，根据系统的承受能力调整可运行线程数量的大小等
package thread_pool;
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;
public class CachedThreadPoolDemo {
public static void main(String[] args) {
//创建线程池对象
ExecutorService executorService = Executors.newCachedThreadPool();
//提交任务
for (int i = 0; i < 20; i++) {
executorService.execute(new MyTask());
}
//关闭资源（线程池）
executorService.shutdown();
}
}
package thread_pool;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class FixedThreadPoolDemo {
public static void main(String[] args) {
ExecutorService executorService = Executors.newFixedThreadPool(5);
for (int i = 0; i < 20; i++) {
executorService.execute(new MyTask());
}
executorService.shutdown();
}
}
线程池的生命周期
RUNNING ：能接受新提交的任务，并且也能处理阻塞队列中的任务；
SHUTDOWN：关闭状态，不再接受新提交的任务，但却可以继续处理阻塞队列中已保存的任务。
STOP：不能接受新任务，也不处理队列中的任务，会中断正在处理任务的线程。
TIDYING：如果所有的任务都已终止了，workerCount (有效线程数) 为0，线程池进入该状态后会调用 terminated() 方法进入TERMINATED 状态。
TERMINATED：在terminated() 方法执行完后进入该状态，默认terminated()方法中什么也没有做
参数说明
corePoolSize：核心线程池的大小
maximumPoolSize：线程池能创建线程的最大个数
keepAliveTime：空闲线程存活时间
unit：时间单位，为keepAliveTime指定时间单位
workQueue：阻塞队列，用于保存任务的阻塞队列
threadFactory：创建线程的工程类
handler：饱和策略（拒绝策略）
public class Consumer implements Runnable {
private BlockingQueue blockingQueue;
public Consumer(BlockingQueue blockingQueue) {
this.blockingQueue = blockingQueue;
}
public void run() {
try {
while(true) {
System.out.println("取出来的元素是："+blockingQueue.take());
}
} catch (Exception e) {
System.out.println("消费者在等待新产品的时候被打断了！");
e.printStackTrace();
}
}
}
public class MainClass {
public static void main(String[] args) {
BlockingQueue blockingQueue = new ArrayBlockingQueue(3,true);
Producer producerPut = new Producer(blockingQueue);
Consumer consumer = new Consumer(blockingQueue);
// ProducerOffer producerOffer = new ProducerOffer(blockingQueue);
new Thread(producerPut).start();
new Thread(consumer).start();
}
}
public class Producer implements Runnable {
private BlockingQueue blockingQueue;
private static int element = 0;
public Producer(BlockingQueue blockingQueue) {
this.blockingQueue = blockingQueue;
}
public void run() {
try {
while(element < 20) {
System.out.println("将要放进去的元素是："+element);
blockingQueue.put(element++);
}
} catch (Exception e) {
System.out.println("生产者在等待空闲空间的时候被打断了！");
e.printStackTrace();
}
System.out.println("生产者已经终止了生产过程！");
}
}
拒绝策略
ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。
ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务
execute方法执行逻辑
如果当前运行的线程少于corePoolSize，则会创建新的线程来执行新的任务；
如果运行的线程个数等于或者大于corePoolSize，则会将提交的任务存放到阻塞队列workQueue中；
如果当前workQueue队列已满的话，则会创建新的线程来执行任务；
如果线程个数已经超过了maximumPoolSize，则会使用饱和策略RejectedExecutionHandler来进行处理。
Executor和Submit
submit是基于方法Executor.execute(Runnable)的延伸，通过创建并返回一个Future类对象可用于取消执行和/或等待完成。
public class DelayQueueTest {
public static void main(String[] args) {
DelayQueue queue = new DelayQueue<>();
queue.add(new DelayTask("1", 1000L, TimeUnit.MILLISECONDS));
queue.add(new DelayTask("2", 1000L, TimeUnit.MILLISECONDS));
queue.add(new DelayTask("3", 1000L, TimeUnit.MILLISECONDS));
System.out.println("queue put done");
while(!queue.isEmpty()) {
try {
DelayTask task = queue.take();
System.out.println(task.name + ":" + System.currentTimeMillis());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
class DelayTask implements Delayed {
public String name;
public Long delayTime;
public TimeUnit delayTimeUnit;
public Long executeTime;//ms
DelayTask(String name, long delayTime, TimeUnit delayTimeUnit) {
this.name = name;
this.delayTime = delayTime;
this.delayTimeUnit = delayTimeUnit;
this.executeTime = System.currentTimeMillis() + delayTimeUnit.toMillis(delayTime);
}
@Override
public int compareTo(Delayed o) {
if(this.getDelay(TimeUnit.MILLISECONDS) > o.getDelay(TimeUnit.MILLISECONDS)) {
return 1;
}else if(this.getDelay(TimeUnit.MILLISECONDS) < o.getDelay(TimeUnit.MILLISECONDS)) {
return -1;
}
return 0;
}
@Override
public long getDelay(TimeUnit unit) {
return unit.convert(executeTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
}
}
public class DelayQueueTest {
public static void main(String[] args) {
DelayQueue queue = new DelayQueue<>();
queue.add(new DelayTask("1", 1000L, TimeUnit.MILLISECONDS));
queue.add(new DelayTask("2", 1000L, TimeUnit.MILLISECONDS));
queue.add(new DelayTask("3", 1000L, TimeUnit.MILLISECONDS));
System.out.println("queue put done");
while(!queue.isEmpty()) {
try {
DelayTask task = queue.take();
System.out.println(task.name + ":" + System.currentTimeMillis());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
class DelayTask implements Delayed {
public String name;
public Long delayTime;
public TimeUnit delayTimeUnit;
public Long executeTime;//ms
DelayTask(String name, long delayTime, TimeUnit delayTimeUnit) {
this.name = name;
this.delayTime = delayTime;
this.delayTimeUnit = delayTimeUnit;
this.executeTime = System.currentTimeMillis() + delayTimeUnit.toMillis(delayTime);
}
@Override
public int compareTo(Delayed o) {
if(this.getDelay(TimeUnit.MILLISECONDS) > o.getDelay(TimeUnit.MILLISECONDS)) {
return 1;
}else if(this.getDelay(TimeUnit.MILLISECONDS) < o.getDelay(TimeUnit.MILLISECONDS)) {
return -1;
}
return 0;
}
@Override
public long getDelay(TimeUnit unit) {
return unit.convert(executeTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
}
}
public class UsePriorityBlockingQueue {
public static void main(String[] args) throws Exception{
PriorityBlockingQueue q = new PriorityBlockingQueue();
Task t1 = new Task();
t1.setId(3);
t1.setName("id为3");
Task t2 = new Task();
t2.setId(4);
t2.setName("id为4");
Task t3 = new Task();
t3.setId(1);
t3.setName("id为1");
q.add(t1); //3
q.add(t2); //4
q.add(t3); //1
System.out.println("容器：" + q);
System.out.println(q.take().getId());
System.out.println("容器：" + q);
}
}
public class Task implements Comparable{ private int id; private String name; public int getId() { return id; } public void setId(int id) { this.id = id; } public String getName() { return name; } public void setName(String name) { this.name = name; } @Override public int compareTo(Task task) { return this.id > task.id ? 1 : (this.id < task.id ? -1 : 0); } public String toString() { return this.id + "," + this.name; } }
public class SynchronousQueueExample {
static class SynchronousQueueProducer implements Runnable {
protected BlockingQueue blockingQueue;
final Random random = new Random();
public SynchronousQueueProducer(BlockingQueue queue) {
this.blockingQueue = queue;
}
@Override
public void run() {
while (true) {
try {
String data = UUID.randomUUID().toString();
System.out.println(Thread.currentThread().getName() + "---Put: " + data);
blockingQueue.put(data);
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
static class SynchronousQueueConsumer implements Runnable {
protected BlockingQueue blockingQueue;
public SynchronousQueueConsumer(BlockingQueue queue) {
this.blockingQueue = queue;
}
@Override
public void run() {
while (true) {
try {
String data = blockingQueue.take();
System.out.println(Thread.currentThread().getName()
+ " take(): " + data);
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
public static void main(String[] args) {
final BlockingQueue synchronousQueue = new SynchronousQueue();
SynchronousQueueProducer queueProducer = new SynchronousQueueProducer(
synchronousQueue);
new Thread(queueProducer).start();
SynchronousQueueConsumer queueConsumer1 = new SynchronousQueueConsumer(
synchronousQueue);
new Thread(queueConsumer1).start();
SynchronousQueueConsumer queueConsumer2 = new SynchronousQueueConsumer(
synchronousQueue);
new Thread(queueConsumer2).start();
}
}
线程池的关闭
关闭线程池，可以通过shutdown和shutdownNow两个方法
原理：遍历线程池中的所有线程，然后依次中断
1、shutdownNow首先将线程池的状态设置为STOP,然后尝试停止所有的正在执行和未执行任务的线程，并返回等待执行任务的列表；
2、shutdown只是将线程池的状态设置为SHUTDOWN状态，然后中断所有没有正在执行任务的线程
