2. 程式人生 > >Boost無鎖佇列

Boost無鎖佇列

阿新 • • 發佈：2019-01-04

在開發接收轉發agent時，採用了多執行緒的生產者-消費者模式，用了加互斥鎖的方式來實現執行緒同步。互斥鎖會阻塞執行緒，所以壓測時，效率並不高。所以想起用無鎖佇列來實現，效能確實提升了。

首先介紹下lock-free和wait-free的區別：

阻塞演算法可能會出現整個系統都掛起的情況（佔有鎖的執行緒被中斷，無法釋放所，那麼所有試圖爭用這個鎖的執行緒會被掛起），系統中的所有執行緒全部餓死。

無鎖演算法可以保證系統中至少有一個執行緒處於工作狀態，但是還是可能有執行緒永遠搶不到資源而被餓死。

無等待演算法保證系統中的所有執行緒都能處於工作狀態，沒有執行緒會被餓死，只要時間夠，所有執行緒都能結束。相比於無鎖演算法，無等待演算法有更強的保證。

一. 用互斥鎖實現單生產者-單消費者

#include <string>
#include <sstream>
#include <list>
#include <pthread.h>
#include <iostream>
#include <time.h>

using namespace std;

int producer_count = 0;
int consumer_count = 0;

list<string> product;
list<string> consumer_list;
pthread_mutex_t mutex;

const
 
 int iterations = 10000;

//是否生產完畢標誌
bool done = false;

void* producer(void* args)
{
    for (int i = 0; i != iterations; ++i) {
        pthread_mutex_lock(&mutex);
        int value = ++producer_count;
        stringstream ss;
        ss<<value;
        product.push_back(ss.str());
        //cout<<"list push:"<<ss.str()<<endl;
 

        pthread_mutex_unlock(&mutex);
    }
    return 0;
}



//消費函式
void* consumer(void* args)
{
    //當沒有生產完畢，則邊消費邊生產
    while (!done) {
        pthread_mutex_lock(&mutex);
        if(!product.empty()){
            consumer_list.splice(consumer_list.end(), product);
            pthread_mutex_unlock(&mutex);
            while(!consumer_list.empty()){
                string value = consumer_list.front();
                consumer_list.pop_front();
                //cout<<"list pop:"<<value<<endl;
                ++consumer_count;
            }           
        }else{
            pthread_mutex_unlock(&mutex);
        }
    }
    //如果生產完畢，則消費
    while(!consumer_list.empty()){
        string value = consumer_list.front();
        consumer_list.pop_front();
        //cout<<"list pop:"<<value<<endl;
        ++consumer_count;
    }
    return 0;
}

int main(int argc, char* argv[])
{
    struct timespec time_start={0, 0},time_end={0, 0};
    clock_gettime(CLOCK_REALTIME, &time_start);

    pthread_t producer_tid;
    pthread_t consumer_tid;

    pthread_mutex_init (&mutex,NULL);
    pthread_create(&producer_tid, NULL, producer, NULL);
    pthread_create(&consumer_tid, NULL, consumer, NULL);

    //等待生產者生產完畢
    pthread_join(producer_tid, NULL);
    //可以消費標誌
    done = true;     //主執行緒不等生產執行緒完畢就設定done標記
    cout << "producer done" << endl;    //輸出以觀察主執行緒和各子執行緒的執行順序

    //等待消費者結束
    pthread_join(consumer_tid, NULL);
    clock_gettime(CLOCK_REALTIME, &time_end);

    long cost = (time_end.tv_sec-time_start.tv_sec)/1000000 + (time_end.tv_nsec-time_start.tv_nsec)/1000;

    cout<<"===========cost time:"<<cost<<"us==========="<<endl;

    cout << "produced " << producer_count << " objects." << endl;
    cout << "consumed " << consumer_count << " objects." << endl;
}

生產消費10000個string型別的資料，耗時：58185us

二. Boost庫的無鎖佇列

boost.lockfree實現了三種無鎖資料結構：
boost::lockfree::queue
alock-free multi-produced/multi-consumer queue
一個無鎖的多生產者/多消費者佇列，注意，這個queue不支援string型別，支援的資料型別要求：
- T must have a copy constructor
- T must have a trivial assignment operator
- T must have a trivial destructor

boost::lockfree::stack
alock-free multi-produced/multi-consumer stack
一個無鎖的多生產者/多消費者棧，支援的資料型別要求：
- T must have a copy constructor

boost::lockfree::spsc_queue
await-free single-producer/single-consumer queue (commonly known as ringbuffer)
一個無等待的單生產者/單消費者佇列（通常被稱為環形緩衝區），支援的資料型別要求：
- T must have a default constructor
- T must be copyable

三. Queue示例

這裡實現的還是單生產者-單消費者。

#include <pthread.h>
#include <boost/lockfree/queue.hpp>
#include <iostream>
#include <time.h>
#include <boost/atomic.hpp>

using namespace std;

//生產數量
boost::atomic_int producer_count(0);
//消費數量
boost::atomic_int consumer_count(0);
//佇列
boost::lockfree::queue<int> queue(512);


//迭代次數
const int iterations = 10000;

//生產函式
void* producer(void* args)
{
    for (int i = 0; i != iterations; ++i) {
        int value = ++producer_count;
        //原子計數————多執行緒不存在計數不上的情況       
        //若沒有進入佇列，則重複推送
        while(!queue.push(value));
        //cout<<"queue push:"<<value<<endl;
    }
    return 0;
}

//是否生產完畢標誌
boost::atomic<bool> done (false);

//消費函式
void* consumer(void* args)
{
    int value;
    //當沒有生產完畢，則邊消費邊生產
    while (!done) {
        //只要能彈出元素，就消費
        while (queue.pop(value)) {
            //cout<<"queue pop:"<<value<<endl;
            ++consumer_count;
        }
    }
    //如果生產完畢，則消費
    while (queue.pop(value)){
        //cout<<"queue pop:"<<value<<endl;
        ++consumer_count;
    }
    return 0;
}

int main(int argc, char* argv[])
{
    cout << "boost::lockfree::queue is ";
    if (!queue.is_lock_free())
        cout << "not ";
    cout << "lockfree" << endl;

    struct timespec time_start={0, 0},time_end={0, 0};
    clock_gettime(CLOCK_REALTIME, &time_start);

    pthread_t producer_tid;
    pthread_t consumer_tid;

    pthread_create(&producer_tid, NULL, producer, NULL);
    pthread_create(&consumer_tid, NULL, consumer, NULL);

    //等待生產者生產完畢
    pthread_join(producer_tid, NULL);
    //可以消費標誌
    done = true;     //主執行緒不等生產執行緒完畢就設定done標記
    cout << "producer done" << endl;    //輸出以觀察主執行緒和各子執行緒的執行順序

    //等待消費者結束
    pthread_join(consumer_tid, NULL);
    clock_gettime(CLOCK_REALTIME, &time_end);

    long cost = (time_end.tv_sec-time_start.tv_sec)/1000000 + (time_end.tv_nsec-time_start.tv_nsec)/1000;

    cout<<"===========cost time:"<<cost<<"us==========="<<endl;

    //輸出生產和消費數量
    cout << "produced " << producer_count << " objects." << endl;
    cout << "consumed " << consumer_count << " objects." << endl;

    return 0;
}

生產消費10000個int型別的資料，耗時：3963us
stack與queue類似，只不過是先進後出。

四. Waitfree Single-Producer/Single-Consumer Queue無等待單生產者/單消費者佇列

#include <pthread.h>
#include <iostream>
#include <time.h>
#include <boost/lockfree/spsc_queue.hpp>
#include <boost/atomic.hpp>

using namespace std;

int producer_count = 0;
boost::atomic_int consumer_count (0);

boost::lockfree::spsc_queue<int, boost::lockfree::capacity<1024> > spsc_queue;

const int iterations = 10000;

void* producer(void* args)
{
    for (int i = 0; i != iterations; ++i) {
        int value = ++producer_count;
        while(!spsc_queue.push(value));
        //cout<<"queue push:"<<value<<endl;
    }
    return 0;
}

//是否生產完畢標誌
boost::atomic<bool> done (false);

//消費函式
void* consumer(void* args)
{
    int value;
    //當沒有生產完畢，則邊消費邊生產
    while (!done) {
        //只要能彈出元素，就消費
        while (spsc_queue.pop(value)) {
            //cout<<"queue pop:"<<value<<endl;
            ++consumer_count;
        }
    }
    //如果生產完畢，則消費
    while (spsc_queue.pop(value)){
        //cout<<"queue pop:"<<value<<endl;
        ++consumer_count;
    }
    return 0;
}

int main(int argc, char* argv[])
{
    using namespace std;
    cout << "boost::lockfree::queue is ";
    if (!spsc_queue.is_lock_free())
        cout << "not ";
    cout << "lockfree" << endl;

    struct timespec time_start={0, 0},time_end={0, 0};
    clock_gettime(CLOCK_REALTIME, &time_start);

    pthread_t producer_tid;
    pthread_t consumer_tid;

    pthread_create(&producer_tid, NULL, producer, NULL);
    pthread_create(&consumer_tid, NULL, consumer, NULL);

    //等待生產者生產完畢
    pthread_join(producer_tid, NULL);
    //可以消費標誌
    done = true;     //主執行緒不等生產執行緒完畢就設定done標記
    cout << "producer done" << endl;    //輸出以觀察主執行緒和各子執行緒的執行順序

    //等待消費者結束
    pthread_join(consumer_tid, NULL);
    clock_gettime(CLOCK_REALTIME, &time_end);

    long cost = (time_end.tv_sec-time_start.tv_sec)/1000000 + (time_end.tv_nsec-time_start.tv_nsec)/1000;

    cout<<"===========cost time:"<<cost<<"us==========="<<endl;

    cout << "produced " << producer_count << " objects." << endl;
    cout << "consumed " << consumer_count << " objects." << endl;
}

生產消費10000個int型別的資料，耗時：1832us
如果把int改為string型別，耗時：28788us

五.效能對比

這裡寫圖片描述
從上面可以看出在單生產者-單消費者模式下，spsc_queue比queue效能好，無鎖佇列比互斥鎖的方式效能也要好。

相關推薦

Boost無鎖佇列

在開發接收轉發agent時，採用了多執行緒的生產者-消費者模式，用了加互斥鎖的方式來實現執行緒同步。互斥鎖會阻塞執行緒，所以壓測時，效率並不高。所以想起用無鎖佇列來實現，效能確實提升了。 首先介紹下lock-free和wait-free的區別： 阻塞演算法可

Boost——無鎖佇列queue

boost提供了三種無鎖方案：boost::lockfree::queue;支援多生產者，多消費者的無鎖佇列；boost::lockfree::stack；支援多生產者，多消費的無鎖棧；boost::lockfree::spsc_queue；支援單生產者，單消費者的無鎖佇列，

evpp效能測試（3）: 對無鎖佇列boost::lockfree::queue和moodycamel::ConcurrentQueue做一個性能對比測試

Brief 我們使用https://github.com/Qihoo360/evpp專案中的EventLoop::QueueInLoop(...)函式來做這個效能測試。我們通過該函式能夠將一個仿函式執行體從一個執行緒排程到另一個執行緒中執行。這是一個典

boost之無鎖佇列使用例項

#ifndef SEARCH_ENGINE_H #define SEARCH_ENGINE_H #include "boost_comm.h" #include "message_header.h" namespace tspace { class search_en

基於陣列的無鎖佇列(譯)

原文 1 引言 最近對於注重效能的應用程式,我們有了一種能顯著提高程式效能的選擇:多執行緒.執行緒的概念實際上已經存在了很長時間.在過去,多數計算機只有一個處理器,執行緒主要用於將一個大的任務拆分成一系列更小的執行單元.以使得當其中某些執行單元因為等待資源而被阻塞的時候剩餘的執行單元能繼續執行。舉個示例,一個

lockFreeQueue 無鎖佇列實現與總結

無鎖佇列 介紹 　　在工程上，為了解決兩個處理器互動速度不一致的問題，我們使用佇列作為快取，生產者將資料放入佇列，消費者從佇列中取出資料。這個時候就會出現四種情況，單生產者單消費者，多生產者單消費者，單生成者多消費者，多生產者多消費者。我們知道，多執行緒往往會帶來資料不一致的情況，一般需要靠加鎖解決問題。

Go語言無鎖佇列元件的實現 (chan/interface/select)

1. 背景 go程式碼中要實現非同步很簡單，go funcName()。 但是程序需要控制協程數量在合理範圍內，對應大批量任務可以使用“協程池 + 無鎖佇列”實現。 2. golang無鎖佇列實現思路 Channel是Go中的一個核心型別，你可以把它看成一個管道，通過它併發核心單元就可以傳送或者接

Disruptor無鎖佇列淺析

    近期在看作業系統相關資料的時候，閱讀到“訊號量與PV操作”，主要分三塊：互斥控制，同步控制，生產者與消費者問題。因為我日常與伺服器及訊息佇列打交道較多，對生產者與消費者問題比較感興趣，正好之前曾經研究過“Disruptor無鎖佇列”的實現原理，正好再結合PV操作重新

深入理解dpdk rte_ring無鎖佇列

同樣用面向物件的思想來理解無鎖佇列ring。dpdk的無鎖佇列ring是借鑑了linux核心kfifo無鎖佇列。ring的實質是FIFO的環形佇列。 ring的特點： 無鎖出入隊（除了cas(compare and swap)操作）多消費/生產者同時出入隊 使用方法： 1.建立一個ring物件。 介面：s

多執行緒無鎖佇列的實現

一、什麼是多執行緒無鎖佇列？ 多執行緒無鎖佇列還是有鎖的，只不過是用了cpu層面的CAS原子操作，用到這個操作，只需要在取佇列元素和新增佇列元素的時候利用CAS原子操作，就可以保證多個執行緒對佇列元素的有序存取； 二、什麼是CAS操作？ CAS = Compare &am

一個用 C++ 實現的快速無鎖佇列

在程序間傳遞資料很煩人，真心煩人。一步做錯，資料就會損壞。（相較於其他讀寫方式）即使資料排列正確，也更易出錯。 一如既往的，有兩種方式處理這個問題：簡單的方式、麻煩的方式。 簡單的方式 使用你使用平臺提供的鎖（互斥、臨界區域，或等效）。這從概念上不難理解，使用上更簡單。你無需擔心排列問題，庫

c語言資料結構應用-陣列佇列（無鎖佇列）在多執行緒中的使用

一、背景 上篇文章《c語言資料結構實現-陣列佇列/環形佇列》講述了陣列佇列的原理與實現，本文編寫一個雙執行緒進行速度測試 二、相關知識 多執行緒程式設計介面: 1) 建立執行緒 pthread_create 函式 SYNOPSIS #include <

無鎖佇列--基於linuxkfifo實現

一直想寫個無鎖的佇列，來提高專案後臺的效率。 偶然看到linux核心的kfifo.h 實現原理。於是自己仿照了這個實現，目前linux應該是可以對外提供介面了。 #ifndef _NO_LOCK_QUEUE_H_ #define _NO_LOCK_QUEUE_H_ #i

併發無鎖佇列學習（單生產者單消費者模型）

1、引言 本文介紹單生產者單消費者模型的佇列。根據寫入佇列的內容是定長還是變長，分為單生產者單消費者定長佇列和單生產者單消費者變長佇列兩種。單生產者單消費者模型的佇列操作過程是不需要進行加鎖的。生產者通過寫索引控制入隊操作，消費者通過讀索引控制出佇列操作。二者

C++無鎖程式設計資料，無鎖佇列等

1. Lamport's Lock-Free Ring Buffer        [Lamport, Comm. of ACM, 1977]      也就常說的單生產者-單消費者 的ringbuffer， 限制就是隻能一個讀執行緒（消費者），一個寫程序（生產者）。

併發無鎖佇列學習之二【單生產者單消費者】

1、前言 　　最近工作比較忙，加班較多，每天晚上回到家10點多了。我不知道自己還能堅持多久，既然選擇了就要做到最好。寫部落格的少了。總覺得少了點什麼，需要繼續學習。今天繼續上個開篇寫，介紹單生產者單消費者模型的佇列。根據寫入佇列的內容是定長還是變長，分為單生產者單消費者定長佇列和單生產者單消費者變長佇列兩

C++ 無鎖佇列 ABA

實驗環境:vs2013  新建一個無stdafx.h預編譯頭的控制檯程式，然後複製以下程式碼 1、連結串列實現無鎖佇列 2、陣列實現無鎖佇列 1、連結串列 注意: Enqueue函式中有使用new分配記憶體，本人在windows下使用VS2013編譯，這裡的new是執行

無鎖佇列的原理與實現

最近幾天在思考無鎖佇列，看了相關文章，也讀了一些部落格，最後寫了一份程式碼，程式碼實現對在多執行緒環境下對佇列的讀和寫是不需要加鎖的。程式碼如下所示： #include <windows.h> #pragma comment(lib, "Kernel32.li

一讀一寫無鎖佇列c++實現

限制一個執行緒讀，一個執行緒寫，不加鎖的佇列，使用單鏈表實現，測試環境：centos 5.9  [[email protected] test]# cat  test.cpp     #include <iostream> #include <

單生產者，單消費者無鎖佇列實現（c）

根據上面連結所說的原理實現的單生產者，單消費者無鎖佇列 bool __sync_bool_compare_and_swap (type *ptr, type oldval，type newval, ...) 函式提供原子的比較和交換，如果*ptr == oldval