并发队列的介绍及使用

在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