阻塞隊列之LinkedBlockingQueue

概述

LinkedBlockingQueue內部由單鏈表實現，只能從head取元素，從tail添加元素。添加元素和獲取元素都有獨立的鎖，也就是說LinkedBlockingQueue是讀寫分離的，讀寫操作可以並行執行。LinkedBlockingQueue采用可重入鎖(ReentrantLock)來保證在並發情況下的線程安全。

構造器

LinkedBlockingQueue一共有三個構造器，分別是無參構造器、可以指定容量的構造器、可以穿入一個容器的構造器。如果在創建實例的時候調用的是無參構造器，LinkedBlockingQueue的默認容量是Integer.MAX_VALUE，這樣做很可能會導致隊列還沒有滿，但是內存卻已經滿了的情況（內存溢出）。

public LinkedBlockingQueue()；   //設置容量為Integer.MAX

public LinkedBlockingQueue(int capacity)；  //設置指定容量

public LinkedBlockingQueue(Collection<? extends E> c)；  //穿入一個容器，如果調用該構造器，容量默認也是Integer.MAX_VALUE

LinkedBlockingQueue常用操作

取數據

take()：首選。當隊列為空時阻塞

poll()：彈出隊頂元素，隊列為空時，返回空

peek()：和poll烈性，返回隊隊頂元素，但頂元素不彈出。隊列為空時返回null

remove(Object o)：移除某個元素，隊列為空時拋出異常。成功移除返回true

 

添加數據

put()：首選。隊滿是阻塞

offer()：隊滿時返回false

 

判斷隊列是否為空

size()方法會遍歷整個隊列，時間復雜度為O(n),所以最好選用isEmtpy

 

put元素原理

基本過程：

1.判斷元素是否為null，為null拋出異常

2.加鎖(可中斷鎖)

3.判斷隊列長度是否到達容量，如果到達一直等待

4.如果沒有隊滿，enqueue()在隊尾加入元素

5.隊列長度加1，此時如果隊列還沒有滿，調用signal喚醒其他堵塞隊列

 if (e == null) throw new NullPointerException();
       
        int c = -1;
        Node<E> node = new Node<E>(e);
        final ReentrantLock putLock = this.putLock;
        final AtomicInteger count = this.count;
        putLock.lockInterruptibly();
        try {
            while (count.get() == capacity) {
                notFull.await();
            }
            enqueue(node);
            c = count.getAndIncrement();
            if (c + 1 < capacity)
                notFull.signal();
        } finally {
            putLock.unlock();
        }

 

take元素原理

 基本過程：

1.加鎖(依舊是ReentrantLock)，注意這里的鎖和寫入是不同的兩把鎖

2.判斷隊列是否為空，如果為空就一直等待

3.通過dequeue方法取得數據

3.取走元素后隊列是否為空，如果不為空喚醒其他等待中的隊列

public E take() throws InterruptedException {
        E x;
        int c = -1;
        final AtomicInteger count = this.count;
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lockInterruptibly();
        try {
            while (count.get() == 0) {
                notEmpty.await();
            }
            x = dequeue();
            c = count.getAndDecrement();
            if (c > 1)
                notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
        if (c == capacity)
            signalNotFull();
        return x;
    }

enqueue()和dequeue()方法實現都比較簡單，無非就是將元素添加到隊尾，從隊頂取走元素，感興趣的朋友可以自己去看一下，這里就不粘貼了。

 

LinkedBlockingQueue與LinkedBlockingDeque比較

 

LinkedBlockingDeque和LinkedBlockingQueue的相同點在於： 
1. 基於鏈表 
2. 容量可選，不設置的話，就是Int的最大值

和LinkedBlockingQueue的不同點在於： 
1. 雙端鏈表和單鏈表 
2. 不存在哨兵節點 
3. 一把鎖+兩個條件

實例：

 小記：AtomicInteger的getAndIncrment和getAndDcrement()等方法，這些方法分為兩步，get和increment(decrement)，在get和increment中間可能有其他線程進入，導致多個線程get到的數值是相同的，也會導致多個線程累加后的值其實累加1.在這種情況下，使用volatile也是沒有效果的，因為get之后沒有對值進行修改，不能觸發volatile的效果。

public class ProducerAndConsumer {
    public static void main(String[] args){

        try{
            BlockingQueue queue = new LinkedBlockingQueue(5);

            ExecutorService executor = Executors.newFixedThreadPool(5);
            Produer producer = new Produer(queue);
            for(int i=0;i<3;i++){
                executor.execute(producer);
            }
            executor.execute(new Consumer(queue));

            executor.shutdown();
        }catch (Exception e){
            e.printStackTrace();
        }

    }
}

class Produer implements  Runnable{

    private BlockingQueue queue;
    private int nums = 20;  //循環次數

    //標記數據編號
    private static volatile AtomicInteger count = new AtomicInteger();
    private boolean isRunning = true;
    public Produer(){}

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

    public void run() {
        String data = null;
        try{
            System.out.println("開始生產數據");
            System.out.println("-----------------------");

          while(nums>0){
                nums--;
                count.decrementAndGet();

                Thread.sleep(500);
                System.out.println(Thread.currentThread().getId()+ " :生產者生產了一個數據");
                queue.put(count.getAndIncrement());
            }
        }catch(Exception e){
            e.printStackTrace();
            Thread.currentThread().interrupt();
        }finally{
            System.out.println("生產者線程退出！");
        }
    }
}

class Consumer implements Runnable{

    private BlockingQueue queue;
    private int nums = 20;
    private boolean isRunning = true;

    public Consumer(){}

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

    public void run() {

        System.out.println("消費者開始消費");
        System.out.println("-------------------------");

        while(nums>0){
            nums--;
            try{
                while(isRunning){
                    int data = (Integer)queue.take();
                    Thread.sleep(500);
                    System.out.println("消費者消費的數據是" + data);
            }

            }catch(Exception e){
                e.printStackTrace();
                Thread.currentThread().interrupt();
            }finally {
                System.out.println("消費者線程退出!");
            }

        }
    }
}

效果：

12 :生產者生產了一個數據
11 :生產者生產了一個數據
13 :生產者生產了一個數據
12 :生產者生產了一個數據
消費者消費的數據是-3
11 :生產者生產了一個數據
13 :生產者生產了一個數據
12 :生產者生產了一個數據
消費者消費的數據是-3
13 :生產者生產了一個數據
11 :生產者生產了一個數據
12 :生產者生產了一個數據
消費者消費的數據是-3
13 :生產者生產了一個數據
11 :生產者生產了一個數據
消費者消費的數據是-3
消費者消費的數據是-3

可以看到，有多個producer在生產數據的時候get到的是相同的值。