并发队列的介绍及使用
在JDK1.5新加入了一个包concurrent，位于java.util.concurrent。在我们写业务代码的时候，可能最为常见就是ConcurrentHashMap。当然今天我们的主角不是他，而是queue。在并发队列上JDK提供了两套实现。
阻塞队列（IO）：以BlockingQueue接口为代表的阻塞队列。
非阻塞（NIO）队列：以ConcurrentLinkedQueue为代表的高性能队列。
1.阻塞队列（IO）
1.1.什么是阻塞队列
1、当队列是满时，往队列里添加元素的操作会被阻塞；
2、或者当队列是空的时，从队列中获取元素的操作将会被阻塞；
3、阻塞队列是线程安全的。
1.2.阻塞队列的应用场景
阻塞队列常用于生产者和消费者的场景。
1、生产者是往队列里添加元素的线程；
2、消费者是从队列里拿元素的线程；
3、阻塞队列就是生产者存放元素的容器。
1.3.BlockingQueue
1.3.1.ArrayBlockingQueue
1、是一个有边界的阻塞队列，它的内部实现是一个数组。有边界的意思是它的容量是有限的，我们必须在其初始化的时候指定它的容量大小，容量大小一旦指定就不可改变。
2、采用FIFO先进先出的方式存储数据，最新插入的对象是尾部，最新移出的对象是头部。
1.3.2.ArrayBlockingQueue
1、LinkedBlockingQueue阻塞队列大小的配置是可选的，如果我们初始化时指定一个大小，它就是有边界的，如果不指定，它就是无边界的。说是无边界，其实是采用了默认大小为Integer.MAX_VALUE的容量 。它的内部实现是一个链表。
2、和ArrayBlockingQueue一样，LinkedBlockingQueue 也是以先进先出的方式存储数据，最新插入的对象是尾部，最新移出的对象是头部。
1.3.3.PriorityBlockingQueue
1、是一个没有边界的队列，它的排序规则和 java.util.PriorityQueue一样
2、需要注意，PriorityBlockingQueue中允许插入null对象。
3、所有插入PriorityBlockingQueue的对象必须实现 java.lang.Comparable接口，队列优先级的排序规则就是按照我们对这个接口的实现来定义的。
4、我们可以从PriorityBlockingQueue获得一个迭代器Iterator，但这个迭代器并不保证按照优先级顺序进行迭代。
1.3.4.SynchronousQueue
仅允许容纳一个元素。当一个线程插入一个元素后会被阻塞，除非这个元素被另一个线程消费。
2.非阻塞（NIO）队列
和阻塞队列相比，他不会被阻塞，并且效率比阻塞高。
2.1.ConcurrentLinkedQueue
ConcurrentLinkedQueue是一个适用于高并发场景下的队列，通过无锁的方式，实现了高并发状态下的高性能。通常ConcurrentLinkedQueue性能好于BlockingQueue。它是一个基于链接节点的*线程安全队列。
该队列的元素遵循先进先出的原则。头是最先加入的，尾是最近加入的，该队列不允许null元素。
2.2.1重要方法
add 和offer() 都是加入元素的方法(在ConcurrentLinkedQueue中这俩个方法没有任何区别)，在这两个方法中，均调用offerLast（E e）方法。
将指定的元素插入此双端队列的末尾
/**
* Inserts the specified element at the tail of this deque.
* As the deque is unbounded, this method will never throw
* {@link IllegalStateException} or return {@code false}.
*
* @return {@code true} (as specified by {@link Collection#add})
* @throws NullPointerException if the specified element is null
*/
public boolean add(E e) {
return offerLast(e);
}
/**
* Inserts the specified element at the tail of this deque.
* As the deque is unbounded, this method will never return {@code false}.
*
* @return {@code true} (as specified by {@link Queue#offer})
* @throws NullPointerException if the specified element is null
*/
public boolean offer(E e) {
return offerLast(e);
}
poll() 和peek() 都是取头元素节点，区别在于前者会删除元素，后者不会。
public E poll() { return pollFirst(); }
public E peek() { return peekFirst(); }
返回删除此双端队列的第一个元素
/**
* Retrieves and removes the first element of this deque,
* or returns {@code null} if this deque is empty.
*
* @return the head of this deque, or {@code null} if this deque is empty
*/
public E pollFirst() {
restart: for (;;) {
for (Node<E> first = first(), p = first;;) {
final E item;
if ((item = p.item) != null) {
// recheck for linearizability
if (first.prev != null) continue restart;
if (ITEM.compareAndSet(p, item, null)) {
unlink(p);
return item;
}
}
if (p == (p = p.next)) continue restart;
if (p == null) {
if (first.prev != null) continue restart;
return null;
}
}
}
}
返回此双端队列的第一个元素
/**
* Retrieves, but does not remove, the first element of this deque,
* or returns {@code null} if this deque is empty.
*
* @return the head of this deque, or {@code null} if this deque is empty
*/
public E peekFirst() {
restart: for (;;) {
E item;
Node<E> first = first(), p = first;
while ((item = p.item) == null) {
if (p == (p = p.next)) continue restart;
if (p == null)
break;
}
// recheck for linearizability
if (first.prev != null) continue restart;
return item;
}
}
2.2.2代码演示
@Test
public void test() throws Exception {
ConcurrentLinkedDeque<String> qStr = new ConcurrentLinkedDeque<>();
qStr.add("1胡涛");
qStr.add("2谢天");
qStr.add("3杨腾龙");
System.out.println("当前队列:"+qStr.toString());
String poll2 = qStr.peek();
System.out.println("调用peek()方法获取到"+poll2);
System.out.println("调用poll()方法后，队列还有："+qStr.toString());
String poll = qStr.poll();
System.out.println("调用poll()方法获取到"+poll);
System.out.println("调用poll()方法后，队列还有："+qStr.toString());
}
3.阻塞队列的生产消费者应用
基础知识储备：你需要多线程的知识点，这里我就不啰嗦多线程的知识点了。
AtomicInteger：高并发的情况下，i++无法保证原子性，往往会出现问题，所以引入AtomicInteger类。
volatile：1、保证共享变量对所有的线程的可见；2、禁止CPU进行重排序
生产者
class ProducerThread implements Runnable {
private BlockingQueue<String> blockingQueue;
private AtomicInteger count = new AtomicInteger();
private volatile boolean FLAG = true;
public ProducerThread(BlockingQueue<String> blockingQueue) {
this.blockingQueue = blockingQueue;
}
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + "生产者开始启动....");
while (FLAG) {
String data = count.incrementAndGet() + "";
try {
boolean offer = blockingQueue.offer(data, 2, TimeUnit.SECONDS);
if (offer) {
System.out.println(Thread.currentThread().getName() + ",生产队列" + data + "成功..");
} else {
System.out.println(Thread.currentThread().getName() + ",生产队列" + data + "失败..");
}
Thread.sleep(1000);
} catch (Exception e) {
}
}
System.out.println(Thread.currentThread().getName() + ",生产者线程停止...");
}
public void stop() {
this.FLAG = false;
}
}
消费者
class ConsumerThread implements Runnable {
private volatile boolean FLAG = true;
private BlockingQueue<String> blockingQueue;
public ConsumerThread(BlockingQueue<String> blockingQueue) {
this.blockingQueue = blockingQueue;
}
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + "消费者开始启动....");
while (FLAG) {
try {
String data = blockingQueue.poll(2, TimeUnit.SECONDS);
if (data == null || data == "") {
FLAG = false;
System.out.println("消费者超过2秒时间未获取到消息.");
return;
}
System.out.println("消费者获取到队列信息成功,data:" + data);
} catch (Exception e) {
// TODO: handle exception
}
}
}
}
@Test
public void proDucerAndConsumer() throws Exception {
BlockingQueue<String> blockingQueue = new LinkedBlockingQueue<>(3);
ProducerThread producerThread = new ProducerThread(blockingQueue);
ConsumerThread consumerThread = new ConsumerThread(blockingQueue);
Thread t1 = new Thread(producerThread);
Thread t2 = new Thread(consumerThread);
t1.start();
t2.start();
//10秒后 停止线程..
try {
Thread.sleep(10*1000);
producerThread.stop();
} catch (Exception e) {
// TODO: handle exception
}
}
运行效果
本文地址：https://blog.csdn.net/m0_37892044/article/details/107160584
推荐阅读