快捷導航

基于條件變量的消息隊列說明介紹

更新時間：2013年04月19日 14:14:00 作者：

本篇文章小編為大家介紹，基于條件變量的消息隊列說明介紹。需要的朋友參考一下

條件變量是線程之前同步的另一種機制。條件變量給多線程提供了一種會和的場所。當條件變量和互斥鎖一起使用時，允許線程以無競爭的方式等待特定的條件發(fā)生。這樣大大減少了鎖競爭引起的線程調(diào)度和線程等待。

消息隊列是服務(wù)器端開發(fā)過程中繞不開的一道坎，前面，我已經(jīng)實現(xiàn)了一個基于互斥鎖和三隊列的消息隊列，性能很不錯。博客園中的其他園主也實現(xiàn)了很多基于環(huán)形隊列和lock-free的消息隊列，很不錯，今天我們將要實現(xiàn)一個基于雙緩沖、互斥鎖和條件變量的消息隊列；這個大概也參考了一下java的blockingqueue，在前面一個博客中有簡單介紹??！基于三緩沖的隊列，雖然最大限度上解除了線程競爭，但是在玩家很少，消息很小的時候，需要添加一些buff去填充數(shù)據(jù)，這大概也是其一個缺陷吧！

消息隊列在服務(wù)器開發(fā)過程中主要用于什么對象呢？

1：我想大概就是通信層和邏輯層之間的交互，通信層接受到的網(wǎng)絡(luò)數(shù)據(jù)，驗證封包之后，通過消息隊列傳遞給邏輯層，邏輯層將處理結(jié)果封包再傳遞給通信層！

2：邏輯線程和數(shù)據(jù)庫IO線程的分離；數(shù)據(jù)庫IO線程負責對數(shù)據(jù)庫的讀寫更新，邏輯層對數(shù)據(jù)庫的操作，封裝成消息去請求數(shù)據(jù)庫IO線程，數(shù)據(jù)庫IO線程處理完之后，再交回給邏輯層。

3：日志；處理模式與方式2 類似。不過日志大概是不需要返回的！

給出源代碼：

BlockingQueue.h文件

復制代碼代碼如下:

/* 
 * BlockingQueue.h 
 * 
 *  Created on: Apr 19, 2013 
 *      Author: archy_yu 
 */

#ifndef BLOCKINGQUEUE_H_ 
#define BLOCKINGQUEUE_H_ 

#include <queue> 
#include <pthread.h> 

typedef void* CommonItem; 

class BlockingQueue 
{ 
public: 
    BlockingQueue(); 

    virtual ~BlockingQueue(); 

    int peek(CommonItem &item); 

    int append(CommonItem item); 

private: 

    pthread_mutex_t _mutex; 

    pthread_cond_t _cond; 

    std::queue<CommonItem> _read_queue; 

    std::queue<CommonItem> _write_queue; 

}; 

  
#endif /* BLOCKINGQUEUE_H_ */

BlockingQueue.cpp 文件代碼

復制代碼代碼如下:

/* 
 * BlockingQueue.cpp 
 * 
 *  Created on: Apr 19, 2013 
 *      Author: archy_yu 
 */

#include "BlockingQueue.h" 

BlockingQueue::BlockingQueue() 
{ 
    pthread_mutex_init(&this->_mutex,NULL); 
    pthread_cond_init(&this->_cond,NULL); 
} 

BlockingQueue::~BlockingQueue() 
{ 
    pthread_mutex_destroy(&this->_mutex); 
    pthread_cond_destroy(&this->_cond); 
} 

int BlockingQueue::peek(CommonItem &item) 
{ 

    if( !this->_read_queue.empty() ) 
    { 
        item = this->_read_queue.front(); 
        this->_read_queue.pop(); 
    } 
    else
    { 
        pthread_mutex_lock(&this->_mutex); 

        while(this->_write_queue.empty()) 
        { 
            pthread_cond_wait(&this->_cond,&this->_mutex); 
        } 

        while(!this->_write_queue.empty()) 
        { 
            this->_read_queue.push(this->_write_queue.front()); 
            this->_write_queue.pop(); 
        } 

        pthread_mutex_unlock(&this->_mutex); 
    } 

  
    return 0; 
} 

int BlockingQueue::append(CommonItem item) 
{ 
    pthread_mutex_lock(&this->_mutex); 
    this->_write_queue.push(item); 
    pthread_cond_signal(&this->_cond); 
    pthread_mutex_unlock(&this->_mutex); 
    return 0; 
}

測試代碼：

復制代碼代碼如下:

BlockingQueue _queue; 

void* process(void* arg) 
{ 

    int i=0; 
    while(true) 
    { 
        int *j = new int(); 
        *j = i; 
        _queue.append((void *)j); 
        i ++; 
    } 
    return NULL; 
} 

int main(int argc,char** argv) 
{ 
    pthread_t pid; 
    pthread_create(&pid,0,process,0); 

    long long int start = get_os_system_time(); 
    int i = 0; 
    while(true) 
    { 
        int* j = NULL; 
        _queue.peek((void* &)j); 

        i ++; 

        if(j != NULL && (*j) == 100000) 
        { 
            long long int end = get_os_system_time(); 
            printf("consume %d\n",end - start); 
            break; 
        } 
    } 

    return 0; 
}