java多线程-03-阻塞队列简介

Mar 15, 2017 | Java-多线程

声明

该系列文章只是记录本人回顾java多线程编程时候记录的笔记。文中所用语言并非严谨的专业术语(太严谨的术语其实本人也不会……)。难免有理解偏差的地方，欢迎指正。
另外，大神请绕路。不喜勿喷。
毕竟好记性不如烂笔头嘛，而且许多东西只要不是你经常用的最终都会一丢丢一丢丢地给忘记。

初看之下，阻塞队列似乎和多线程没有多大关系。但是平时使用阻塞队列的场景往往是和线程相关的。所以，此处将阻塞队列也归入java多线程的分类。

1 什么是阻塞队列

说白了，阻塞队列还是一个队列(FIFO)。至于队列的概念就不多说了。
阻塞队列和普通队列的区别就在阻塞这个词。

阻塞的意思，可以这么理解：

当尝试队列出队操作时，若队列已经是空的，则调用出队操作的线程将被阻塞，直到队列有可用元素为止
当尝试队列入队操作时，若队列已经是满的，则调用入队操作的线程将被阻塞，直到队列不再是满的为止

2 JDK提供的阻塞队列

2.1 JDK内置的阻塞队列

ArrayBlockingQueue: 数组结构的有界阻塞队列
LinkedBlockingQueue: 链表结构的有界阻塞队列
LinkedTransferQueue: 链表结构的无界阻塞队列
LinkedBlockingDeque: 链表结构的双向阻塞队列
PriorityBlockingQueue: 支持按优先级排序的无界阻塞队列
DelayQueue: 使用优先级队列(PriorityQueue)实现的阻塞队列，优先队列的比较基准值是时间
SynchronousQueue: 不知道怎么描述这个队列,因为他并不真正存储元素。每个插入操作必须等待另一个线程的移除操作，同样一个移除操作都等待另一个线程的插入操作。

2.2 阻塞队列不可用时的处理方式

此处要说的是，队列在特殊情况下的出队入队操作的处理。
比如在队列为空或者队列已经满了的时候不同的阻塞队列实现或者不同的方法会有不同的处理方式。

对这种特殊情况的处理，大致有以下一些方法:

抛异常
- 队列为空时进行出队操作,可能会有NoSuchElementException
- 队列已满时进行入队操作,可能会有IllegalStateException
返回特殊值
- 抛出NoSuchElementException,可以返回null代替表示队列为空
- 抛出IllegalStateException,可以返回false表示入队失败
一直阻塞
- 队列为空时进行出队操作,进行出队操作的线程将被阻塞,直至队列可用
- 队列已满时进行入队操作,进行入队操作的线程将被阻塞,直至队列可用
超时退出
- 队列为空时进行出队操作,进行出队操作的线程将被阻塞,直至超时退出(线程也会退出)
- 队列已满时进行入队操作,进行入队操作的线程将被阻塞,直至超时退出(线程也会退出)

以下图片是来自《Java并发编程的艺术》一书中的总结:

这队列特殊情况的处理

3 使用示例

以上介绍的七种JDK内置的阻塞队列，各有各的使用场景，一篇文章很难介绍清楚。
以后会抽空补全各种阻塞队列的使用场景示例。

此处，就修改一下上篇文章线程通信中那个生产者和消费者的示例程序(http://blog.csdn.net/hylexus/article/details/53446711)。

和上篇文章的不同之处在于，此处使用阻塞队列来代替上篇文章中简单实现的一个数据结构栈。
Container类的实现修改成了下面的样子:

public static class Container {
	private ArrayBlockingQueue<Product> products;
	public Container(int size) {
		this.products = new ArrayBlockingQueue<>(size);
	}
	public void put(Product product) {
		try {
			this.products.put(product);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
	public Product get() {
		try {
			return this.products.take();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		return null;
	}
}

完整的代码示例如下:

import java.util.concurrent.ArrayBlockingQueue;
public class ProducerConsumerByLockingQueue {
	public static void main(String[] args) {
		Container container = new Container(5);
		for (int i = 0; i < 10; i++) {
			new Thread(new Producer(container), "P-" + i).start();
			new Thread(new Consumer(container), "C-" + i).start();
		}
	}
	public static class Product {
		private String name;
		public Product(String name) {
			super();
			this.name = name;
		}
		public String getName() {
			return name;
		}
		public void setName(String name) {
			this.name = name;
		}
		@Override
		public String toString() {
			return "[name=" + name + "]";
		}
	}
	public static class Container {
		private ArrayBlockingQueue<Product> products;
		public Container(int size) {
			this.products = new ArrayBlockingQueue<>(size);
		}
		public void put(Product product) {
			try {
				this.products.put(product);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		public Product get() {
			try {
				return this.products.take();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			return null;
		}
	}
	public static class Producer implements Runnable {
		private Container container;
		public Producer(Container container) {
			super();
			this.container = container;
		}
		@Override
		public void run() {
			for (int i = 0; i < Integer.MAX_VALUE; i++) {
				Product p = new Product(Thread.currentThread().getName() + "_" + i);
				this.container.put(p);
				System.out.println("生产者[" + Thread.currentThread().getName() + "]生产>>>>>>:" + p);
				try {
					Thread.sleep((long) (Math.random() * 1000));
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
	}
	public static class Consumer implements Runnable {
		private Container container;
		public Consumer(Container container) {
			super();
			this.container = container;
		}
		@Override
		public void run() {
			for (int i = 0; i < Integer.MAX_VALUE; i++) {
				Product product = this.container.get();
				System.out.println("消费者[" + Thread.currentThread().getName() + "]消费<<<<<<:" + product);
				try {
					Thread.sleep((long) (Math.random() * 2000));
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
	}
}

参考资料

《java并发编程的艺术》