①.阻塞队列:从名字可以看出,它也是队列的一种,那么它肯定是一个先进先出FIFO的数据结构。与普通队列不同的是,他支持两个附加操作,即阻塞添加和阻塞删除方法
②. 线程1往阻塞队列中添加元素,而线程2从阻塞队列中移除元素。而在这一系列操作必须符合以下规定:
public interface BlockingQueue<E> extends Queue<E> {
boolean add(E e);
boolean offer(E e);
void put(E e) throws InterruptedException;
boolean offer(E e, long timeout, TimeUnit unit)throws InterruptedException;
E take() throws InterruptedException;
E poll(long timeout, TimeUnit unit)throws InterruptedException;
int remainingCapacity();
boolean remove(Object o);
public boolean contains(Object o);
int drainTo(Collection<? super E> c);
int drainTo(Collection<? super E> c, int maxElements);
}
public class BlockingQueueTest {
public static void main(String[] args) {
BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<String>(3);
System.out.println("--------以下为add的相关操作---------");
addRemoveTest(blockingQueue);
}
public static void addRemoveTest(BlockingQueue<String> blockingQueue) {
System.out.println("添加状态+\t"+blockingQueue.add("1"));
System.out.println("添加状态+\t"+blockingQueue.add("2"));
System.out.println("添加状态+\t"+blockingQueue.add("3"));
// System.out.println("添加状态+\t"+blockingQueue.add("4"));
System.out.println("队首元素+\t"+blockingQueue.element());
System.out.println("删除元素+\t"+blockingQueue.remove());
System.out.println("队首元素+\t"+blockingQueue.element());
System.out.println("删除元素+\t"+blockingQueue.remove());
System.out.println("队首元素+\t"+blockingQueue.element());
System.out.println("删除元素+\t"+blockingQueue.remove());
// System.out.println("队首元素+\t"+blockingQueue.element());
// System.out.println("删除元素+\t"+blockingQueue.remove(blockingQueue.element()));
}
}
// 未打开注释代码输出如下:
--------以下为add的相关操作---------
添加状态+ true
添加状态+ true
添加状态+ true
队首元素+ 1
删除元素+ 1
队首元素+ 2
删除元素+ 2
队首元素+ 3
删除元素+ 3
// 当队列已满,继续添加元素时打开注释代码,输出如下:
--------以下为add的相关操作---------
添加状态+ true
添加状态+ true
添加状态+ true
Exception in thread "main" java.lang.IllegalStateException: Queue full
at java.util.AbstractQueue.add(AbstractQueue.java:98)
public class BlockingQueueTest {
public static void main(String[] args) {
BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<String>(3);
System.out.println("--------以下为offer的相关操作---------");
offerPollTest(blockingQueue);
}
private static void offerPollTest(BlockingQueue<String> blockingQueue) {
System.out.println("添加状态+\t"+blockingQueue.offer("1"));
System.out.println("添加状态+\t"+blockingQueue.offer("2"));
System.out.println("添加状态+\t"+blockingQueue.offer("3"));
System.out.println("添加状态+\t"+blockingQueue.offer("4"));
System.out.println("队首元素+\t"+blockingQueue.peek());
System.out.println("删除元素+\t"+blockingQueue.poll());
System.out.println("队首元素+\t"+blockingQueue.peek());
System.out.println("删除元素+\t"+blockingQueue.poll());
System.out.println("队首元素+\t"+blockingQueue.peek());
System.out.println("删除元素+\t"+blockingQueue.poll());
System.out.println("删除元素+\t"+blockingQueue.poll());
}
}
// 输出如下:
--------以下为offer的相关操作---------
添加状态+ true
添加状态+ true
添加状态+ true
添加状态+ false
队首元素+ 1
删除元素+ 1
队首元素+ 2
删除元素+ 2
队首元素+ 3
删除元素+ 3
删除元素+ null
// 注意:当队列没有元素的时候使用poll,返回null
public class BlockingQueueTest {
public static void main(String[] args) {
BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<String>(3);
System.out.println("--------以下为put的相关操作---------");
putTakeTest(blockingQueue);
}
private static void putTakeTest(BlockingQueue<String> blockingQueue) {
try{
blockingQueue.put("1");
blockingQueue.put("2");
blockingQueue.put("3");
// blockingQueue.put("4");
System.out.println("删除元素+\t"+blockingQueue.take());
blockingQueue.put("4");
System.out.println("删除元素+\t"+blockingQueue.take());
System.out.println("删除元素+\t"+blockingQueue.take());
System.out.println("删除元素+\t"+blockingQueue.take());
}catch(Exception e){
e.getStackTrace();
}
}
}
// 打开注释代码输出如下:(程序未停止)
--------以下为put的相关操作---------
// 未打开注释代码输出如下:
--------以下为put的相关操作---------
删除元素+ 1
删除元素+ 2
删除元素+ 3
删除元素+ 4
②. ArrayBlockingQueue: 由数组结构组成的有界阻塞队列
③. LinkedBlockingQueue: 由链表结构组成的有界(但大小默认值Integer>MAX_VAL UE)阻塞队列
需要注意的是LinkedBlockingQueue虽然是有界的,但有个巨坑,其默认大小是IntegerMAX_VALUE,高达21亿,一般情况下内存早爆了在线程池的ThreadPoolExecutor有体现
④. SynchronousQueue:不存储元素的阻塞队列,也即是单个元素的队列
public class SynchronusQueueTest {
public static void main(String[] args) {
BlockingQueue<String> synchronusQueue = new SynchronousQueue<>();
new Thread(() ->{
try{
System.out.println(Thread.currentThread().getName()+"\t put 1");
synchronusQueue.put("1");
System.out.println(Thread.currentThread().getName()+"\t put 2");
synchronusQueue.put("2");
System.out.println(Thread.currentThread().getName()+"\t put 3");
synchronusQueue.put("3");
}catch(Exception e){
e.getStackTrace();
}
},"Prod").start();
new Thread(() ->{
try {
try{ TimeUnit.SECONDS.sleep(3); }catch (InterruptedException e){ e.printStackTrace(); }
System.out.println(Thread.currentThread().getName()+"\t take "+synchronusQueue.take());
try{ TimeUnit.SECONDS.sleep(3); }catch (InterruptedException e){ e.printStackTrace(); }
System.out.println(Thread.currentThread().getName()+"\t take "+synchronusQueue.take());
try{ TimeUnit.SECONDS.sleep(3); }catch (InterruptedException e){ e.printStackTrace(); }
System.out.println(Thread.currentThread().getName()+"\t take"+synchronusQueue.take());
} catch (Exception e) {
e.printStackTrace();
}
},"Cons").start();
}
// 输出如下:有时间间隔
Prod put 1
Cons take 1
Prod put 2
Cons take 2
Prod put 3
Cons take3
⑤. PriorityBlockingQueue:支持优先级排序的无界阻塞队列
⑥. LinkedTransferQueue:由链表结构组成的无界阻塞队列
⑦. LinkedBlockingDeque:由了解结构组成的双向阻塞队列
④. 为什么说LinkedBlockingQueue的吞吐性是高于arrayBlockingQueue?
吞吐性能强是因为有两个锁,试想一下,Array里面使用的是一个锁,不管put还是take行为,都可能被这个锁卡住,而Linked里面put和take是两个锁,put只会被put行为卡住,而不会被take卡住,因此吞吐性能自然强于Array。 而“less predictable performance”这个也是显而易见的,Array采用的时固定内存,而Linked采用的时动态内存,无论是分配内存还是释放内存甚至GC动态内存的性能自然都会比固定内存要差
⑤. 锁机制不一样
arrayBlockingQueue使用的一个锁来控制,LinkedBlockingQueue使用了2个锁来控制,一个名为putLock,另一个是takeLock,但是锁的本质都是ReentrantLock
①. 我们需要根据自己的场景、并发情况来评估线程池的几个核心参数,包括核心线程数、最大线程数、线程回收策略、工作队列的类型,以及拒绝策略,确保线程池的工作行为符合需求,一般都需要设置有界的工作队列和可控的线程数
②. 任何时候,都应该为自定义线程池指定有意义的名称,以方便排查问题。当出现线程数量暴增、线程死锁、线程占用大量 CPU、线程执行出现异常等问题时,我们往往会抓取线程栈。此时,有意义的线程名称,就可以方便我们定位问题
③. newFixedThreadPool方法的源码不难发现,线程池的工作队列直接new了一个LinkedBlockingQueue,而默认构造方法的 LinkedBlockingQueue是一个Integer.MAX_VALUE 长度的队列,可以认为是无界的
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
public class LinkedBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
...
/**
* Creates a {@code LinkedBlockingQueue} with a capacity of
* {@link Integer#MAX_VALUE}.
*/
public LinkedBlockingQueue() {
this(Integer.MAX_VALUE);
}
...
}
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());