阻塞队列

是什么？

顾名思义,首先它是一个队列
当阻塞队列是空时,从队列中获取元素的操作将会被阻塞
当阻塞队列是满时,往队列中添加元素的操作将会被阻塞

为什么用?有什么好处?

好处是我们不需要关心什么时候需要阻塞线程,什么时候需要唤醒线程,因为BlockingQueue都一手给你包办好了

在concurrent包发布以前,在多线程环境下,我们每个程序员都必须自己去控制这些细节,尤其还要兼顾效率和线程安全,而这会给我们的程序带来不小的复杂度.

种类

①. ArrayBlockingQueue: 由数组结构组成的有界阻塞队列

②. LinkedBlockingQueue: 由链表结构组成的有界(但大小默认值 Integer>MAX_VAL UE)阻塞队列.

③. SynchronousQueue:不存储元素的阻塞队列,也即是单个元素的队列。也就是说SynchronousQueue的每一次insert操作，必须等待其他线性的remove操作。而每一个remove操作也必须等待其他线程的insert操作。

public class BlockingQueueExceptionDemo {public static void main(String[] args) {//List list=new ArrayList();//注意:这里要给一个初始值BlockingQueueblockingQueue=new ArrayBlockingQueue<>(3);//add() 方法是给ArrayBlockingQueue添加元素,如果超过会抛出异常!System.out.println(blockingQueue.add("a"));//trueSystem.out.println(blockingQueue.add("b"));//trueSystem.out.println(blockingQueue.add("c"));//true//element是检查元素有没有? 检查的是出栈的元素blockingQueue.element();//remove()System.out.println(blockingQueue.remove());//aSystem.out.println(blockingQueue.remove());//bSystem.out.println(blockingQueue.remove());//c}
}@Testpublic void offerAndPoll()throws Exception{BlockingQueue blockingQueue=new ArrayBlockingQueue(3);//前三个直接成功System.out.println(blockingQueue.offer("a", 2L, TimeUnit.SECONDS));System.out.println(blockingQueue.offer("a", 2L, TimeUnit.SECONDS));System.out.println(blockingQueue.offer("a", 2L, TimeUnit.SECONDS));//下面这个会阻塞2sSystem.out.println(blockingQueue.offer("a", 2L, TimeUnit.SECONDS));}

线程池

为什么使用线程池,优势?

1. 线程池做的工作主要是控制运行的线程的数量,处理过程中将任务加入队列,然后在线程创建后启动这些任务,如果显示超过了最大数量,超出的数量的线程排队等候,等其他线程执行完毕,再从队列中取出任务来执行
2. 它的主要特点为:线程复用 | 控制最大并发数 | 管理线程.

代码展示：

public class ExecutorTest {public static void main(String[] args) {System.out.println(Runtime.getRuntime().availableProcessors());ExecutorService threadPool1= Executors.newFixedThreadPool(5);//一池5个处理线程ExecutorService threadPool2=Executors.newSingleThreadExecutor();//一池一线程ExecutorService threadPool= Executors.newCachedThreadPool();//一池N线程try {for (int i = 1; i <= 10; i++) {//使用threadPool.execute(() -> {//模拟10个用户来办理业务,每个用户就是一个来自外部的请求线程System.out.println(Thread.currentThread().getName() + "\t 办理业务~！");});//try { TimeUnit.SECONDS.sleep(3);  } catch (InterruptedException e) {e.printStackTrace();}}}catch (Exception e){}finally {//关闭threadPool.shutdown();}}
}

线程池的七大参数

①.corePoolSize:线程池中的常驻核心线程数
在创建了线程池后,当有请求任务来之后,就会安排池中的线程去执行请求任务,近似理解为今日当值线程
当线程池中的线程数目达到corePoolSize后,就会把到达的任务放入到缓存队列当中.
②. maximumPoolSize:线程池能够容纳同时执行的最大线程数,此值大于等于1

③. keepAliveTime:多余的空闲线程存活时间,当空间时间达到keepAliveTime值时,多余的线程会被销毁直到只剩下corePoolSize个线程为止(非核心线程)

④. unit:keepAliveTime的单位

⑤. workQueue:任务队列,被提交但尚未被执行的任务(候客区)

⑥. threadFactory:表示生成线程池中工作线程的线程工厂,用户创建新线程,一般用默认即可(银行网站的logo | 工作人员的制服 | 胸卡等)

⑦. handler:拒绝策略,表示当线程队列满了并且工作线程大于等于线程池的最大显示 数(maxnumPoolSize)时如何来拒绝

线程池的底层工作原理?


拒绝策略

①. 等待队列也已经排满了,再也塞不下新的任务了。同时,线程池的maximumPoolSize也到达了,无法接续为新任务服务,这时我们需要拒绝策略机制合理的处理这个问题

②. JDK内置的拒绝策略

AbortPolicy(默认):直接抛出RejectedException异常阻止系统正常运行
CallerRunsPolicy:"调用者运行"一种调节机制,该策略既不会抛弃任务,也不会抛出异常,而是返回给调用者进行处理
DiscardOldestPolicy:将最早进入队列的任务删除,之后再尝试加入队列
DiscardPolicy:直接丢弃任务,不予任何处理也不抛出异常.如果允许任务丢失,这是最好的拒绝策略
③. 以上内置策略均实现了RejectExecutionHandler接口

线程池参数怎么设置合适
1、cpu密集型

2、io密集型

生产者消费者

Synchronized

public class ThreadWaitNotifyDemo {public static void main(String[] args) {AirCondition airCondition=new AirCondition();new Thread(()->{ for (int i = 1; i <11 ; i++) airCondition.increment();},"线程A").start();new Thread(()->{ for (int i = 1; i <11 ; i++) airCondition.decrement();},"线程B").start();new Thread(()->{ for (int i = 1; i <11 ; i++) airCondition.increment();},"线程C").start();new Thread(()->{ for (int i = 1; i <11 ; i++) airCondition.decrement();},"线程D").start();}
}
class AirCondition{private int number=0;public synchronized void increment(){//1.判断，不能用if，因为多生产者消费者会出现虚假唤醒while(number!=0){try {this.wait();} catch (InterruptedException e) {e.printStackTrace();}}//2.干活number++;System.out.println(Thread.currentThread().getName()+":"+number);//3.唤醒this.notifyAll();}public  synchronized void decrement(){while (number==0){try {this.wait();} catch (InterruptedException e) {e.printStackTrace();}}number--;System.out.println(Thread.currentThread().getName()+":"+number);this.notifyAll();}
}

ReentrantLock

/** 使用Lock代替Synchronized来实现新版的生产者和消费者模式 !* */
@SuppressWarnings("all")
public class ThreadWaitNotifyDemo {public static void main(String[] args) {AirCondition airCondition=new AirCondition();new Thread(()->{ for (int i = 0; i <10 ; i++) airCondition.decrement();},"线程A").start();new Thread(()->{ for (int i = 0; i <10 ; i++) airCondition.increment();},"线程B").start();new Thread(()->{ for (int i = 0; i <10 ; i++) airCondition.decrement();},"线程C").start();new Thread(()->{ for (int i = 0; i <10 ; i++) airCondition.increment();},"线程D").start();}
}
class AirCondition{private int number=0;//定义Lock锁对象final Lock lock=new ReentrantLock();final Condition condition  = lock.newCondition();//生产者,如果number=0就 number++public  void increment(){lock.lock();try {//1.判断while(number!=0){try {condition.await();//this.wait();} catch (InterruptedException e) {e.printStackTrace();}}//2.干活number++;System.out.println(Thread.currentThread().getName()+":\t"+number);//3.唤醒condition.signalAll();//this.notifyAll();}catch (Exception e){e.printStackTrace();}finally {lock.unlock();}}//消费者,如果number=1,就 number--public   void decrement(){lock.lock();try {//1.判断while(number==0){try {condition.await();//this.wait();} catch (InterruptedException e) {e.printStackTrace();}}//2.干活number--;System.out.println(Thread.currentThread().getName()+":\t"+number);//3.唤醒condition.signalAll();//this.notifyAll();}catch (Exception e){e.printStackTrace();}finally {lock.unlock();}}
}

精确通知

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;/*多个线程之间按顺序调用,实现A->B->C
三个线程启动,要求如下:AA打印5次,BB打印10次,CC打印15次接着AA打印5次,BB打印10次,CC打印15次....来10轮
* */
public class ThreadOrderAccess {public static void main(String[] args) {ShareResource shareResource = new ShareResource();new Thread(() -> {for (int i = 1; i <= 10; i++) shareResource.print5();}, "线程A").start();new Thread(() -> {for (int i = 1; i <= 10; i++) shareResource.print10();}, "线程B").start();new Thread(() -> {for (int i = 1; i <= 10; i++) shareResource.print15();}, "线程C").start();}
}class ShareResource {//设置一个标识,如果是number=1,线程A执行...private int number = 1;Lock lock = new ReentrantLock();Condition condition1 = lock.newCondition();Condition condition2 = lock.newCondition();Condition condition3 = lock.newCondition();public void print5() {lock.lock();try {//1.判断while (number != 1) {condition1.await();}//2.干活for (int i = 1; i <= 5; i++) {System.out.println(Thread.currentThread().getName() + ":\t" + i);}//3.唤醒number = 2;condition2.signal();} catch (Exception e) {e.printStackTrace();} finally {lock.unlock();}}public void print10() {lock.lock();try {//1.判断while (number != 2) {condition2.await();}//2.干活for (int i = 1; i <= 10; i++) {System.out.println(Thread.currentThread().getName() + ":\t" + i);}//3.唤醒number = 3;condition3.signal();} catch (Exception e) {e.printStackTrace();} finally {lock.unlock();}}public void print15() {lock.lock();try {//1.判断while (number != 3) {condition3.await();}//2.干活for (int i = 1; i <= 15; i++) {System.out.println(Thread.currentThread().getName() + ":\t" + i);}//3.唤醒number = 1;condition1.signal();} catch (Exception e) {e.printStackTrace();} finally {lock.unlock();}}
}

LockSupport

什么是LockSupport?

①. 通过park()和unpark(thread)方法来实现阻塞和唤醒线程的操作
②. LockSupport是一个线程阻塞工具类,所有的方法都是静态方法,可以让线程在任意位置阻塞,阻塞之后也有对应的唤醒方法。归根结底,LockSupport调用的Unsafe中的native代码
③. 官网解释:
LockSupport是用来创建锁和其他同步类的基本线程阻塞原语
LockSupport类使用了一种名为Permit(许可)的概念来做到阻塞和唤醒线程的功能,每个线程都有一个许可(permit),permit只有两个值1和零,默认是零
可以把许可看成是一种(0,1)信号量(Semaphore),但与Semaphore不同的是,许可的累加上限是1

LockSupport它的解决的痛点
①. LockSupport不用持有锁块,不用加锁,程序性能好

②. 先后顺序,不容易导致卡死(因为unpark获得了一个凭证,之后再调用park方法,就可以名正言顺的凭证消费,故不会阻塞)

③. 代码演示:

/*
(1).阻塞(permit默认是O,所以一开始调用park()方法,当前线程就会阻塞,直到别的线程将当前线程的permit设置为1时,park方法会被唤醒,然后会将permit再次设置为O并返回)static void park()static void park(Object blocker)
(2).唤醒
static void unpark(Thread thread)(调用unpark(thread)方法后,就会将thread线程的许可permit设置成1(注意多次调用unpark方法,不会累加,permit值还是1)会自动唤醒thread线程,即之前阻塞中的LockSupport.park()方法会立即返回)static void unpark(Thread thread)
* */
public class LockSupportDemo {public static void main(String[] args) {Thread t1=new Thread(()->{System.out.println(Thread.currentThread().getName()+"\t"+"coming....");LockSupport.park();/*如果这里有两个LockSupport.park(),因为permit的值为1,上一行已经使用了permit所以下一行被注释的打开会导致程序处于一直等待的状态* *///LockSupport.park();System.out.println(Thread.currentThread().getName()+"\t"+"被B唤醒了");},"A");t1.start();//下面代码注释是为了A线程先执行//try { TimeUnit.SECONDS.sleep(3);  } catch (InterruptedException e) {e.printStackTrace();}Thread t2=new Thread(()->{System.out.println(Thread.currentThread().getName()+"\t"+"唤醒A线程");//有两个LockSupport.unpark(t1),由于permit的值最大为1,所以只能给park一个通行证LockSupport.unpark(t1);//LockSupport.unpark(t1);},"B");t2.start();}
}