java queue队列demo

软件设计

2017-06-16

关注关注

1.ArrayBlockingQueue：基于数组实现的一个阻塞队列，在创建ArrayBlockingQueue对象时必须制定容量大小。

　并且可以指定公平性与非公平性，默认情况下为非公平的，即不保证等待时间最长的队列最优先能够访问队列。

2.ArrayBlockingQueue内部通过Object[]数组保存数据的，也就是说ArrayBlockingQueue本质上是通过数组实现的。

ArrayBlockingQueue的大小，即数组的容量是在创建创建ArrayBlockingQueue时候指定的。

3.如下图所示，ArrayBlockingQueue和ReentrantLock是组合关系，ArrayBlockingQueue中包含一个ReentrantLock对象。

ReentrantLock是可重入的互斥锁。ArrayBlockingQueue就是根据ReentrantLock互斥锁实现"多线程对共享资源的访问"。

ReentrantLock分为公平锁和非公平锁，关于具体使用公平锁还是非公平锁，在创建ArrayBlockingQueue时可以指定；

而且，ArrayBlockingQueue默认会使用非公平锁。

4.ArrayBlockingQueue和Condition是组合关系，ArrayBlockingQueue中包含两个Condition对象(notEmpty和notFull)。

使用通知模式实现:所谓通知模式，当生产者往满的队列里面添加元素的时候，会阻塞生产者（调用Condition notFull.await()进行等待）；

当消费者消费了一个队列中的元素后，会通知（调用Condition notFull.signal()唤醒生产者)生产者当前队列可用。

反之，当消费者消费的时候，发现队列是空的，则消费者会被阻塞（通过Condition的 notEmpty.await()进行等待），

当生产者插入了队列中的一个元素后，则会调用notEmpty.signal()唤醒消费者继续消费。

package com.demo.zlearn.queue;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class ArrayBlockingQueueDemo {
    public static void main(String[] args) throws Exception {

        BlockingQueue queue = new ArrayBlockingQueue(5,true);
        Producer producer = new Producer(queue);
        Consumer consumer1 = new Consumer(queue,1000);
        new Thread(producer).start();
        new Thread(consumer1).start();

    }
}
class Producer implements Runnable{

    protected BlockingQueue queue = null;

    public Producer(BlockingQueue queue) {
        this.queue = queue;
    }

    public void run() {
        try {
            for(int i=0;i<10;i++){
                System.out.println("put:"+i);
                queue.put("AA");
                Thread.sleep(4000);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
class Consumer implements Runnable{

    protected BlockingQueue queue = null;

    private long sleepTimes = 0;

    public Consumer(BlockingQueue queue,long sleepTimes) {
        this.queue = queue;
        this.sleepTimes = sleepTimes;
    }

    public void run() {
        try {
            while(true){
                System.out.println("isEmpty:"+queue.isEmpty()+"\n"+
                                   "size:"+queue.size()+"\n"+
                                   "remainingCapacity:"+queue.remainingCapacity()+"\n" +
                                    "take:"+queue.take()+"\n"+
                                    "sleepTimes:"+sleepTimes);  //take 如果队列为空，则阻塞
                System.out.println("-----------------------");

                Thread.sleep(sleepTimes);

            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}----------------------------------------------------------------------------------------------------------------------

put:0isEmpty:truesize:0remainingCapacity:5take:AAsleepTimes:1000-----------------------put:1isEmpty:truesize:0remainingCapacity:5take:AAsleepTimes:1000-----------------------put:2isEmpty:truesize:0remainingCapacity:5take:AAsleepTimes:1000-----------------------put:3isEmpty:truesize:0remainingCapacity:5take:AAsleepTimes:1000-----------------------