c++鏈表實現(xiàn)的阻塞隊列

最近從java源碼里發(fā)現(xiàn)了阻塞隊列的實現(xiàn)，覺得非常有趣。

首先，介紹下什么是阻塞隊列。阻塞隊列代表著一個隊列可以線程安全的往該隊列中寫數(shù)據(jù)和從該隊列中讀數(shù)據(jù)。也就是說，我們可以在多個線程之間并發(fā)的進行寫數(shù)據(jù)和讀數(shù)據(jù)，而不會引發(fā)任何并發(fā)問題。

下面我們就說說如何實現(xiàn)一個阻塞隊列。

而實現(xiàn)一個阻塞隊列的前提：

1. 需要能夠使用鏈表實現(xiàn)一個隊列
2. 能夠使用c++的鎖機制，去給隊列的寫和讀操作加鎖。

為了性能，這里的讀和寫的鎖不能是同一把鎖。而對于一個鏈表隊列來說，讀取操作肯定需要涉及頭指針，寫操作肯定涉及尾指針。既然要實現(xiàn)讀操作一把鎖和寫操作一把鎖。那么就要求讀操作只能更改頭指針而不能更改尾指針，寫操作只能更改尾指針而不能更改頭指針。不滿足這個要求，那么讀寫操作就不可能實現(xiàn)用兩把鎖分別對讀寫進行加鎖。

基本隊列的實現(xiàn)

現(xiàn)在我們先說說如何實現(xiàn)這個隊列。

要求：入隊操作(enqueue)只能操作尾指針(last), 出隊操作(dequeue)只能操作頭指針(head)。

對于隊列的初始化，這里不能單純的設置為空指針，需要將頭尾指針同一節(jié)點。

下面我們來看入隊操作如何實現(xiàn)

從這個入隊操作來看，該操作只更改了尾指針last, 而沒有更改頭指針head。

其代碼實現(xiàn)為：

 void enqueue(int item) {
    Node *node = new Node(item);
    last = last->next = node;
  }

接下來我們來看出隊操作如何實現(xiàn)

從出隊操作來看就有趣的多，它拋棄了head所指向的節(jié)點，而這個節(jié)點有可能是那些節(jié)點呢？

• 初始化時所賦值的那個節(jié)點
• 出隊后的節(jié)點

也就是說，head所指向的節(jié)點中的val值沒有任何實際含義。當需要出隊時，出隊head指向的下一個節(jié)點first中val的值，然后拋棄head本身指向的值，讓head指向head的下一個節(jié)點first，此時head原來所指向的節(jié)點將被刪除?，F(xiàn)在我們可以看出出隊操作也只改變了頭指針head的值。

其代碼實現(xiàn)為：

 int dequeue() {
    Node *h = head;
    Node *first = head->next;
    delete h;
    head = first;
    int x = first->item;
    return x;
  }

現(xiàn)在隊列已經(jīng)實現(xiàn)，下面就看看阻塞隊列如何實現(xiàn)。

阻塞隊列的實現(xiàn)

既然是阻塞隊列，那就意味這加鎖和等待。那就需要對c++的一些鎖知識和條件變量有了解。

先來看看我們需要實現(xiàn)阻塞隊列的那些方法：

入隊

讓我們先來實現(xiàn)put這個方法。

先看其實現(xiàn)流程圖：

由于該隊列實現(xiàn)有最大值限制，故在插入數(shù)據(jù)之前需要先判斷該隊列是否已滿，已滿則需等待該隊列有可用空間。在該線程入隊操作完成后，可能有別的線程也在等待入隊，需要喚醒其他寫數(shù)據(jù)的線程，使其繼續(xù)執(zhí)行后續(xù)操作。如果入隊前隊列為空，可能有出隊操作正在阻塞等待讀數(shù)，也需要去喚醒讀數(shù)據(jù)的線程。

在看代碼實現(xiàn)之前，我們需要定義一些變量用于代碼實現(xiàn)環(huán)節(jié)：

 /* The capacity bound*/
  int64_t capacity;
  /*Current number of elements */
  std::atomic<int64_t> count;  
	/** Lock held by take, pool, etc */
  std::mutex takeLock;
  /** Wait queue for waiting takes */
  std::condition_variable notEmpty;
  /** Lock held by put, offer, etc */
  std::mutex putLock;
  /** Wait queue for waiting puts */
  std::condition_variable notFull;

put函數(shù)的代碼實現(xiàn)：

void LinkedBlockingQueue::put(const int item){
  int c;
  {
    std::unique_lock<std::mutex> lck{putLock};
    if( count == capacity) {
      notFull.wait(lck, [this](){return count < capacity;}); //（1）
    }
    enqueue(item); //不應該把申請空間放在鎖里面，耗時有點大
    c = count.fetch_add(1);
  }
  if(c + 1 < capacity) {
    notFull.notify_one();
  }

  if(0 == c) {
    notEmpty.notify_one();
  }
}

對于offer(o)的實現(xiàn)，主要更改是對上述代碼(1)中的等待改為直接返回false, 表示，當前沒有可用空間插入數(shù)據(jù)。正常插入就返回true.

對于offer(o,timeout)的實現(xiàn)，就需要在上述代碼(1)中的wait函數(shù)添加上時間參數(shù)，使其可以在timeout時間內(nèi)返回，如果是正常喚醒，正常入隊，則返回ture,否則返回false.

該更改為：

 if(!notFull.wait_for(lck, rel_time, [this](){return count < capacity;})){
        return false;
      }

出隊

對于出隊，其實現(xiàn)和入隊基本相同，基本上只需要更改其中的關鍵判斷和通知。

take函數(shù)的代碼實現(xiàn)為：

void LinkedBlockingQueue::take(int & returnVal) {
  int c;
  {
    std::unique_lock<std::mutex> lck{takeLock};
    if( 0 == count) {
      notEmpty.wait(lck, [this](){return count > 0 ;});
    }
    returnVal = dequeue();
    c = count.fetch_sub(1);
  }

  if( c > 1 ) {
    notEmpty.notify_one();
  }

  if(c == capacity) {
    notFull.notify_one();
  }
}

剩下的實現(xiàn)細節(jié)可以參考我的實現(xiàn)

到此這篇關于c++實現(xiàn)LinkBlockedQueue的問題的文章就介紹到這了,更多相關c++實現(xiàn)LinkBlockedQueue內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

最新評論