Java -- 使用阻塞队列（BlockingQueue）控制线程通信

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

BlockingQueeu接口是Queue的子接口，但是它的主要作用并不是作为容器，而是作为线程同步的工具。

特征：

当生产者线程试图向BlockingQueue中放入元素时，如果该队列已满，则该线程被阻塞；当消费者线程试图从BlockingQueue中取出元素时，如果该队列已空，则该线程被阻塞。

程序的两个线程通过交替从BlockingQueue中放入元素、取出元素，即可很好地控制线程的通信。

BlockingQueue提供如下两个支持阻塞的方法：

put(E e):尝试从BlockingQueue的头部取出元素，如果该队列的元素已满，则阻塞该线程。

take()：尝试从BlockingQueue的头部取出元素，如果该队列的元素已空，则阻塞该线程。

BlockingQueue继承了Queue接口，当然也可以使用Queue接口的方法。这些方法归纳起来可分为3组：

在队列尾部插入元素。包括add(E e)、offer(E e)和put(E e)方法，当该队列已满时，这3个方法分别抛出异常、返回false、阻塞队列。
在队列头部删除并返回删除的元素。包括remove()、poll()和take()方法。当该队列已空时，这3个方法分别抛出异常、返回false、阻塞队列。
在队列头部取出但是不删除元素。包括element()和peek()方法，当队列已空时，这两个方法分别抛出异常、返回false。

BlockingQueue队列的5个实现类：

ArrayBlockingQueue:基于数组实现的BlockingQueue队列
LinkedBlockingQueue:基于链表实现的BlockingQueue队列
PriorityBlockingQueue:.....
SynchronousQueue:同步队列。对对垒的存、取操作必须交替进行
DelayQueue:.....

具体代码示例：

package com.fuchenggang.blockingqueue;import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;/**
* @ClassName: Producer
* @Description: 生产者
* @author fucg@kunyitech.com
* @date 2015年7月10日下午4:08:03
*/
class Producer extends Thread {
private BlockingQueue<String> bq; public Producer(BlockingQueue<String> bq) {
this.bq = bq;
} @Override
public void run() {
String[] strArr = new String[]
{
"Java",
"Struts",
"Spring"
}; for (int i = 0; i < 999999999; i++) {
System.out.println(getName() + "生产者准备生产集合元素！"); try {
Thread.sleep(200);
//尝试放入元素，如果队列已满，则线程被阻塞
bq.put(strArr[i%3]);
} catch (InterruptedException e) {
e.printStackTrace();
} System.out.println(getName() + "生产完成：" +bq);
}
}
}/**
* @ClassName: Consumer
* @Description: 消费者
* @author fucg@kunyitech.com
* @date 2015年7月10日下午4:08:19
*/
class Consumer extends Thread{ private BlockingQueue<String> bq; public Consumer(BlockingQueue<String> bq) {
this.bq = bq;
} @Override
public void run() {
while (true) {
System.out.println(getName() + "消费者准备消费集合元素"); try {
Thread.sleep(200);
//尝试取出元素，如果队列已空，则线程被阻塞
bq.take();
} catch (InterruptedException e) {
e.printStackTrace();
} System.out.println(getName() + "消费完成：" + bq);
}
}
}/**
* @ClassName: BlockingQueueTest2
* @Description: 测试生产者线程和消费者线程的通信
* @author fucg@kunyitech.com
* @date 2015年7月10日下午4:13:37
*/
public class BlockingQueueTest2 {
public static void main(String[] args) {
//创建一个容器为1的BlockingQueue
BlockingQueue<String> bq = new ArrayBlockingQueue<String>(1); //启动3个生产者线程
new Producer(bq).start();
new Producer(bq).start();
new Producer(bq).start(); //启动1个消费者线程
new Consumer(bq).start();
}
}