1. 程式人生 > >【Java併發之】BlockingQueue

【Java併發之】BlockingQueue

本文主要講的是併發包中涉及到的集合,關於普通集合,請參考【java 集合概覽

一、什麼是BlockingQueue

BlockingQueue即阻塞佇列,從阻塞這個詞可以看出,在某些情況下對阻塞佇列的訪問可能會造成阻塞。被阻塞的情況主要有如下兩種:

1. 當佇列滿了的時候進行入佇列操作
2. 當佇列空了的時候進行出佇列操作

因此,當一個執行緒試圖對一個已經滿了的佇列進行入佇列操作時,它將會被阻塞,除非有另一個執行緒做了出佇列操作;同樣,當一個執行緒試圖對一個空佇列進行出佇列操作時,它將會被阻塞,除非有另一個執行緒進行了入佇列操作。

在Java中,BlockingQueue的介面位於java.util.concurrent

包中(在Java5版本開始提供),由上面介紹的阻塞佇列的特性可知,阻塞佇列是執行緒安全的。

二、BlockingQueue的用法

阻塞佇列主要用在生產者/消費者的場景,下面這幅圖展示了一個執行緒生產、一個執行緒消費的場景:

這裡寫圖片描述

負責生產的執行緒不斷的製造新物件並插入到阻塞佇列中,直到達到這個佇列的上限值。佇列達到上限值之後生產執行緒將會被阻塞,直到消費的執行緒對這個佇列進行消費。同理,負責消費的執行緒不斷的從佇列中消費物件,直到這個佇列為空,當佇列為空時,消費執行緒將會被阻塞,除非佇列中有新的物件被插入。

三、BlockingQueue介面中的方法

阻塞佇列一共有四套方法分別用來進行insert

removeexamine,當每套方法對應的操作不能馬上執行時會有不同的反應,下面這個表格就分類列出了這些方法:
asdf

- Throws Exception Special Value Blocks Times Out
Insert add(o) offer(o) put(o) offer(o, timeout, timeunit)
Remove remove(o) poll() take() poll(timeout, timeunit)
Examine element() peek()

這四套方法對應的特點分別是:

1. ThrowsException:如果操作不能馬上進行,則丟擲異常
2. SpecialValue:如果操作不能馬上進行,將會返回一個特殊的值,一般是true或者false
3. Blocks:如果操作不能馬上進行,操作會被阻塞
4. TimesOut:如果操作不能馬上進行,操作會被阻塞指定的時間,如果指定時間沒執行,則返回一個特殊值,一般是true或者false

需要注意的是,我們不能向BlockingQueue中插入null,否則會報NullPointerException

四、BlockingQueue的實現類

BlockingQueue只是java.util.concurrent包中的一個介面,而在具體使用時,我們用到的是它的實現類,當然這些實現類也位於java.util.concurrent包中。在Java6中,BlockingQueue的實現類主要有以下幾種:

1. ArrayBlockingQueue
2. DelayQueue
3. LinkedBlockingQueue
4. PriorityBlockingQueue
5. SynchronousQueue

下面我們就分別介紹這幾個實現類。

4.1 ArrayBlockingQueue

ArrayBlockingQueue是一個有邊界的阻塞佇列,它的內部實現是一個數組。有邊界的意思是它的容量是有限的,我們必須在其初始化的時候指定它的容量大小,容量大小一旦指定就不可改變。

ArrayBlockingQueue是以先進先出的方式儲存資料,最新插入的物件是尾部,最新移出的物件是頭部。下面是一個初始化和使用ArrayBlockingQueue的例子:

BlockingQueue queue = new ArrayBlockingQueue(1024);
queue.put("1");
Object object = queue.take();

4.2 DelayQueue

DelayQueue阻塞的是其內部元素,DelayQueue中的元素必須實現 java.util.concurrent.Delayed介面,這個介面的定義非常簡單:

public interface Delayed extends Comparable<Delayed> {
long getDelay(TimeUnit unit);
}

getDelay()方法的返回值就是佇列元素被釋放前的保持時間,如果返回0或者一個負值,就意味著該元素已經到期需要被釋放,此時DelayedQueue會通過其take()方法釋放此物件。

從上面Delayed 介面定義可以看到,它還繼承了Comparable介面,這是因為DelayedQueue中的元素需要進行排序,一般情況,我們都是按元素過期時間的優先順序進行排序。

例1:為一個物件指定過期時間

首先,我們先定義一個元素,這個元素要實現Delayed介面

public class DelayedElement implements Delayed {
  private long expired;
  private long delay;
  private String name;

  DelayedElement(String elementName, long delay) {
         this. name = elementName;
         this. delay= delay;
         expired = ( delay + System. currentTimeMillis());
  }

  @Override
  public int compareTo(Delayed o) {
        DelayedElement cached=(DelayedElement) o;
         return cached.getExpired()> expired?1:-1;
  }

  @Override
  public long getDelay(TimeUnit unit) {

         return ( expired - System. currentTimeMillis());
  }

  @Override
  public String toString() {
         return "DelayedElement [delay=" + delay + ", name=" + name + "]";
  }

  public long getExpired() {
         return expired;
  }

}

設定這個元素的過期時間為3s

public class DelayQueueExample {
  public static void main(String[] args) throws InterruptedException {
        DelayQueue<DelayedElement> queue= new DelayQueue<>();
        DelayedElement ele= new DelayedElement( "cache 3 seconds",3000);
         queue.put( ele);
        System. out.println( queue.take());

  }

}

執行這個main函式,我們可以發現,我們需要等待3s之後才會列印這個物件。

其實DelayQueue應用場景很多,比如定時關閉連線、快取物件,超時處理等各種場景,下面我們就拿學生考試為例讓大家更深入的理解DelayQueue的使用。

例2:把所有考試的學生看做是一個DelayQueue,誰先做完題目釋放誰

首先,我們構造一個學生物件

public class Student implements Runnable,Delayed{
  private String name;  //姓名
  private long costTime;//做試題的時間
  private long finishedTime;//完成時間

  public Student(String name, long costTime) {
         this. name = name;
         this. costTime= costTime;
         finishedTime = costTime + System. currentTimeMillis();
  }

  @Override
  public void run() {
        System. out.println( name + " 交卷,用時" + costTime /1000);
  }

  @Override
  public long getDelay(TimeUnit unit) {
         return ( finishedTime - System. currentTimeMillis());
  }

  @Override
  public int compareTo(Delayed o) {
        Student other = (Student) o;
         return costTime >= other. costTime?1:-1;
  }

}

然後在構造一個教師物件對學生進行考試

public class Teacher {
  static final int STUDENT_SIZE = 30;
  public static void main(String[] args) throws InterruptedException {
        Random r = new Random();
        //把所有學生看做一個延遲佇列
        DelayQueue<Student> students = new DelayQueue<Student>();
        //構造一個執行緒池用來讓學生們“做作業”
        ExecutorService exec = Executors.newFixedThreadPool(STUDENT_SIZE);
         for ( int i = 0; i < STUDENT_SIZE; i++) {
               //初始化學生的姓名和做題時間
               students.put( new Student( "學生" + (i + 1), 3000 + r.nextInt(10000)));
        }
        //開始做題
        while(! students.isEmpty()){
               exec.execute( students.take());
        }
         exec.shutdown();
  }
}

我們看一下執行結果:

學生2 交卷,用時3
學生1 交卷,用時5
學生5 交卷,用時7
學生4 交卷,用時8
學生3 交卷,用時11

通過執行結果我們可以發現,每個學生在指定開始時間到達之後就會“交卷”(取決於getDelay()方法),並且是先做完的先交卷(取決於compareTo()方法)。

通過檢視其原始碼可以看到,DelayQueue內部實現用的是PriorityQueue和一個Lock:

這裡寫圖片描述

4.3 LinkedBlockingQueue

LinkedBlockingQueue阻塞佇列大小的配置是可選的,如果我們初始化時指定一個大小,它就是有邊界的,如果不指定,它就是無邊界的。說是無邊界,其實是採用了預設大小為Integer.MAX_VALUE的容量 。它的內部實現是一個連結串列。

和ArrayBlockingQueue一樣,LinkedBlockingQueue 也是以先進先出的方式儲存資料,最新插入的物件是尾部,最新移出的物件是頭部。下面是一個初始化和使LinkedBlockingQueue的例子:

BlockingQueue<String> unbounded = new LinkedBlockingQueue<String>();
BlockingQueue<String> bounded   = new LinkedBlockingQueue<String>(1024);
bounded.put("Value");
String value = bounded.take();

4.4 PriorityBlockingQueue

PriorityBlockingQueue是一個沒有邊界的佇列,它的排序規則和 java.util.PriorityQueue一樣。需要注意,PriorityBlockingQueue中允許插入null物件。

所有插入PriorityBlockingQueue的物件必須實現 java.lang.Comparable介面,佇列優先順序的排序規則就是按照我們對這個介面的實現來定義的。

另外,我們可以從PriorityBlockingQueue獲得一個迭代器Iterator,但這個迭代器並不保證按照優先順序順序進行迭代。

下面我們舉個例子來說明一下,首先我們定義一個物件型別,這個物件需要實現Comparable介面:

public class PriorityElement implements Comparable<PriorityElement> {
private int priority;//定義優先順序
PriorityElement(int priority) {
    //初始化優先順序
    this.priority = priority;
}
@Override
public int compareTo(PriorityElement o) {
    //按照優先順序大小進行排序
    return priority >= o.getPriority() ? 1 : -1;
}
public int getPriority() {
    return priority;
}
public void setPriority(int priority) {
    this.priority = priority;
}
@Override
public String toString() {
    return "PriorityElement [priority=" + priority + "]";
}
}

然後我們把這些元素隨機設定優先順序放入佇列中

public class PriorityBlockingQueueExample {
public static void main(String[] args) throws InterruptedException {
    PriorityBlockingQueue<PriorityElement> queue = new PriorityBlockingQueue<>();
    for (int i = 0; i < 5; i++) {
        Random random=new Random();
        PriorityElement ele = new PriorityElement(random.nextInt(10));
        queue.put(ele);
    }
    while(!queue.isEmpty()){
        System.out.println(queue.take());
    }
}
}

看一下執行結果:

PriorityElement [priority=3]
PriorityElement [priority=4]
PriorityElement [priority=5]
PriorityElement [priority=8]
PriorityElement [priority=9]

4.5 SynchronousQueue

SynchronousQueue佇列內部僅允許容納一個元素。當一個執行緒插入一個元素後會被阻塞,除非這個元素被另一個執行緒消費。