Java核心知识点学习----多线程中的阻塞队列,ArrayBlockingQueue介绍

1.什么是阻塞队列?

所谓队列,遵循的是先进先出原则(FIFO),阻塞队列,即是数据共享时,A在写数据时,B想读同一数据,那么就将发生阻塞了.

看一下线程的四种状态,首先是新创建一个线程,然后,通过start方法启动线程--->线程变为可运行可执行状态,然后通过数据产生共享,线程产生互斥---->线程状态变为阻塞状态---->阻塞状态想打开的话可以调用notify方法.

这里Java5中提供了封装好的类,可以直接调用然后构造阻塞状态,以保证数据的原子性.

2.如何实现?

主要是实现BlockingQueue接口.

比较常见的实现有:ArrayBlockingQueue,LinkedBlockingQueue,DelayedWorkQueue等竺

这里简单介绍ArrayBlockingQueue;

//方式1
new ArrayBlockingQueue(int capacity);//Parameters:capacity the capacity of this queue//方式2
public ArrayBlockingQueue(int capacity, boolean fair) ;//fair if true then queue accesses for threads blocked on insertion or removal, are processed in FIFO order; if false the access order is unspecified.//方式3public ArrayBlockingQueue(int capacity, boolean fair,Collection<? extends E> c) ;//c the collection of elements to initially contain

三种构造函数,比较常用的是1,2两种,2比1只是多了要不要排序,如果排序,那就是FIFO原则,即先进先出.

3.举例:

package com.amos.concurrent;import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;/**
* @ClassName: BlockingQueueTest
* @Description: Java5新特性,阻塞队列
* @author: amosli
* @email:hi_amos@outlook.com
* @date Apr 27, 2014 10:01:51 PM
*/
public class BlockingQueueTest {public static void main(String[] args) {final BlockingQueue queue = new ArrayBlockingQueue(3);for(int i=0;i<2;i++){new Thread(){public void run(){while(true){try {Thread.sleep((long)(Math.random()*1000));System.out.println(Thread.currentThread().getName() + "准备放数据!");                             queue.put(1);System.out.println(Thread.currentThread().getName() + "已经放了数据，" +                             "队列目前有" + queue.size() + "个数据");} catch (InterruptedException e) {e.printStackTrace();}}}}.start();}new Thread(){public void run(){while(true){try {//将此处的睡眠时间分别改为100和1000，观察运行结果Thread.sleep(100);System.out.println(Thread.currentThread().getName() + "准备取数据!");queue.take();System.out.println(Thread.currentThread().getName() + "已经取走数据，" +                             "队列目前有" + queue.size() + "个数据");                    } catch (InterruptedException e) {e.printStackTrace();}}}}.start();            }
}

效果如下图所示:

说明:这里新建一个队列后,主要调用的是put和take两种方法,一个是存,一个是取,这里由于将取的间隔时间设置的比较短,所以基本队列就没放满过.

4.改写之前的代码

方法1:Java核心知识点学习----多线程并发之线程间的通信,notify,wait

方法2:Java核心知识点学习----使用Condition控制线程通信

方法3:使用ArrayBlockingQueue

package com.amos.concurrent;import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;/**
* @ClassName: BlockingQueueCondition
* @Description: 在前面用Condition实现的同步通知的例子的基础上，改为用阻塞队列来实现。
第一个线程：A.take()……..B.put()
第二个线程：B.take()……..A.put()
* @author: amosli
* @email:hi_amos@outlook.com
* @date Apr 28, 2014 12:57:51 AM
*/
public class BlockingQueueCondition {public static void main(String[] args) {ExecutorService service = Executors.newSingleThreadExecutor();final Business3 business = new Business3();service.execute(new Runnable(){public void run() {for(int i=0;i<2;i++){business.sub();}}});for(int i=0;i<3;i++){business.main();}}}class Business3{BlockingQueue subQueue = new ArrayBlockingQueue(1);BlockingQueue mainQueue = new ArrayBlockingQueue(1);{try {mainQueue.put(1);} catch (InterruptedException e) {e.printStackTrace();}}public void sub(){try{mainQueue.take();for(int i=0;i<10;i++){System.out.println(Thread.currentThread().getName() + " : " + i);}subQueue.put(1);}catch(Exception e){}}public void main(){try{subQueue.take();for(int i=0;i<5;i++){System.out.println(Thread.currentThread().getName() + " : " + i);}mainQueue.put(1);}catch(Exception e){}        }
}

功能的实现都是完全一样的,不同的是,使用ArrayBlockingQueue会更简单.

5.官方---代码示例

典型的生产者消费者模型:

package com.amos.concurrent;import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;/**
* @ClassName: BlockingSample
* @Description: 生产者消费者模型,阻塞,只有生产好了,才能去消费
* @author: amosli
* @email:hi_amos@outlook.com
* @date Apr 28, 2014 1:44:07 AM
*/
public class BlockingSample {public static void main(String[] args) {new BlockingSample().new Setup().main();}class Producer implements Runnable {private final BlockingQueue queue;Producer(BlockingQueue q) { queue = q; }public void run() {try {while (true) { queue.put(produce());System.out.println(Thread.currentThread().getName()+" 现在正在生产!");}} catch (Exception ex) { ex.printStackTrace();}}String produce() { System.out.println("produce now ....");return "produce"; }}class Consumer implements Runnable {private final BlockingQueue queue;Consumer(BlockingQueue q) { queue = q; }public void run() {try {while (true) { consume(queue.take()); System.out.println(Thread.currentThread().getName()+" 现在正在消费!");}} catch (InterruptedException ex) { ex.printStackTrace();}}void consume(Object x) {System.out.println("consume...");}}class Setup {void main() {BlockingQueue q = new ArrayBlockingQueue<String>(1);Producer p = new Producer(q);Consumer c1 = new Consumer(q);Consumer c2 = new Consumer(q);new Thread(p).start();new Thread(c1).start();new Thread(c2).start();}}}

效果如下图所示:

由图上可以看出,只有生产了才能消费,否则会形成阻塞.