Java多线程——4 阻塞队列

阻塞队列是Java5线程新特征中的内容，Java定义了阻塞队列的接口java.util.concurrent.BlockingQueue，阻塞队列的概念是:一个指定长度的队列，如果队列满了，添加新元素的操作会被阻塞等待，直到有空位为止。同样，当队列为空时候，请求队列元素的操作同样会阻塞等待，直到有可用元素为止。是一种常用的并发数据结构，常用于生产者-消费者模式。



下面我们看一下阻塞队列类结构



使用BlockingQueue的时候，尽量不要使用Queue继承下来的方法，否则就失去了Blocking的特性了。

在BlockingQueue中，要使用put和take，而非offer和poll。如果要使用offer和poll，也是要使用带等待时间参数的offer和poll。

使用drainTo批量获得其中的内容，能够减少锁的次数。

在Java中，有很多种阻塞队列(BlockingQueue子类)：

ArrayBlockingQueue（最常用）

LinkedBlockingQueue

SynchronousQueue

PriorityBlockingQueue

CompletionService(BlockingQueue+ Executor)

TransferQueue(JDK 7中更快的SynchronousQueue)

一个简单阻塞队列模拟

BlockingQueue2类

public class BlockingQ2
{
private Object notEmpty = new Object();
private Object notFull = new Object();
private Queue<Object> linkedList = new LinkedList<Object>();
private int maxLength = 10;

public Object take() throws InterruptedException
{
//分别需要对notEmpty和notFull加锁
synchronized (notEmpty)
{
if (linkedList.size() == 0)
{
notEmpty.wait();
}
synchronized (notFull)
{
if (linkedList.size() != maxLength)
{
notFull.notifyAll();
}
return linkedList.poll();
}
}
}

public void offer(Object object) throws InterruptedException
{
synchronized (notEmpty)
{
if (linkedList.size() != 0)
{
notEmpty.notifyAll();
}
synchronized (notFull)
{
if (linkedList.size() == maxLength)
{
notFull.wait();
}
linkedList.add(object);
}
}
}
}

MainTest2类

public class MainTest2
{
public static void main(String args[])
{
BlockingQ2 blockingQ = new BlockingQ2();

Thread thread1 = new MyThread3(blockingQ,"thread1");
Thread thread2 = new MyThread4(blockingQ,"thread1");

thread1.start();
thread2.start();

}

}
class MyThread3 extends Thread
{
private BlockingQ2 blockingQ;

public MyThread3(BlockingQ2 blockingQ,String name)
{
super(name);
this.blockingQ = blockingQ;
}

public void run()
{
for (int i = 0; i < 8; i++)
{
try
{
this.sleep(300);
System.out.println("取出元素： " + blockingQ.take());
}
catch (InterruptedException e)
{
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
}

class MyThread4 extends Thread
{
private BlockingQ2 blockingQ;

public MyThread4(BlockingQ2 blockingQ,String name)
{
super(name);
this.blockingQ = blockingQ;
}

public void run()
{
for (int i = 0; i < 10; i++)
{
try
{
this.sleep(200);
}
catch (InterruptedException e)
{
// TODO Auto-generated catch block
e.printStackTrace();
}
try
{
blockingQ.offer(i);
System.out.println("插入元素： " + i);
}
catch (InterruptedException e)
{
e.printStackTrace();
}

}

}
}

运行结果

插入元素： 0

取出元素： 0

插入元素： 1

取出元素： 1

插入元素： 2

插入元素： 3

取出元素： 2

插入元素： 4

取出元素： 3

插入元素： 5

插入元素： 6

取出元素： 4

插入元素： 7

取出元素： 5

插入元素： 8

插入元素： 9

取出元素： 6

取出元素： 7

一个用BlockingQueue实现的生产者-消费者模式;

BlockingQueue4类

public class BlockingQ4
{
private BlockingQueue<Integer> queue = new ArrayBlockingQueue<Integer>(10,true);

public Integer take() throws InterruptedException
{
return queue.take();
}
public void put(Integer i) throws InterruptedException
{
queue.put(i);
}
}

MainTest4类

public class MainTest4
{

public static void main(String[] args)
{
BlockingQ4 blockingQ4 = new BlockingQ4();

Thread thread7 = new Thread7(blockingQ4);
Thread thread8 = new Thread8(blockingQ4);

thread7.start();
thread8.start();
}
}

class Thread7 extends Thread
{
private BlockingQ4 blockingQ4;

public Thread7(BlockingQ4 blockingQ4)
{
this.blockingQ4 = blockingQ4;
}

public void run()
{
try
{
for (int i = 0; i < 8; i++)
{
System.out.println("tack " + blockingQ4.take());
}
}
catch (InterruptedException e)
{
// TODO Auto-generated catch block
e.printStackTrace();
}

}
}

class Thread8 extends Thread
{
private BlockingQ4 blockingQ4;

public Thread8(BlockingQ4 blockingQ4)
{
this.blockingQ4 = blockingQ4;
}

public void run()
{
try
{
for (int i = 0; i < 4; i++)
{
blockingQ4.put(i);
System.out.println("put " + i);
}
}
catch (InterruptedException e)
{
// TODO Auto-generated catch block
e.printStackTrace();
}

}
}

运行结果



程序中是take8次，可见程序在第五次取的时候发生了阻塞