設(shè)計(jì)并發(fā)隊(duì)列
復(fù)制代碼 代碼如下:
#include pthread.h>
#include list>
using namespace std;
template typename T>
class Queue
{
public:
Queue( )
{
pthread_mutex_init(_lock, NULL);
}
~Queue( )
{
pthread_mutex_destroy(_lock);
}
void push(const T data);
T pop( );
private:
listT> _list;
pthread_mutex_t _lock;
};
template typename T>
void QueueT>::push(const T value )
{
pthread_mutex_lock(_lock);
_list.push_back(value);
pthread_mutex_unlock(_lock);
}
template typename T>
T QueueT>::pop( )
{
if (_list.empty( ))
{
throw "element not found";
}
pthread_mutex_lock(_lock);
T _temp = _list.front( );
_list.pop_front( );
pthread_mutex_unlock(_lock);
return _temp;
}
上述代碼是有效的。但是，請考慮這樣的情況：您有一個(gè)很長的隊(duì)列（可能包含超過 100,000 個(gè)元素），而且在代碼執(zhí)行期間的某個(gè)時(shí)候，從隊(duì)列中讀取數(shù)據(jù)的線程遠(yuǎn)遠(yuǎn)多于添加數(shù)據(jù)的線程。因?yàn)樘砑雍腿〕鰯?shù)據(jù)操作使用相同的互斥鎖，所以讀取數(shù)據(jù)的速度會影響寫數(shù)據(jù)的線程訪問鎖。那么，使用兩個(gè)鎖怎么樣？一個(gè)鎖用于讀取操作，另一個(gè)用于寫操作。給出修改后的 Queue 類。
復(fù)制代碼 代碼如下:
template typename T>
class Queue
{
public:
Queue( )
{
pthread_mutex_init(_rlock, NULL);
pthread_mutex_init(_wlock, NULL);
}
~Queue( )
{
pthread_mutex_destroy(_rlock);
pthread_mutex_destroy(_wlock);
}
void push(const T data);
T pop( );
private:
listT> _list;
pthread_mutex_t _rlock, _wlock;
};
template typename T>
void QueueT>::push(const T value )
{
pthread_mutex_lock(_wlock);
_list.push_back(value);
pthread_mutex_unlock(_wlock);
}
template typename T>
T QueueT>::pop( )
{
if (_list.empty( ))
{
throw "element not found";
}
pthread_mutex_lock(_rlock);
T _temp = _list.front( );
_list.pop_front( );
pthread_mutex_unlock(_rlock);
return _temp;
}
設(shè)計(jì)并發(fā)阻塞隊(duì)列
目前，如果讀線程試圖從沒有數(shù)據(jù)的隊(duì)列讀取數(shù)據(jù)，僅僅會拋出異常并繼續(xù)執(zhí)行。但是，這種做法不總是我們想要的，讀線程很可能希望等待（即阻塞自身），直到有數(shù)據(jù)可用時(shí)為止。這種隊(duì)列稱為阻塞的隊(duì)列。如何讓讀線程在發(fā)現(xiàn)隊(duì)列是空的之后等待？一種做法是定期輪詢隊(duì)列。但是，因?yàn)檫@種做法不保證隊(duì)列中有數(shù)據(jù)可用，它可能會導(dǎo)致浪費(fèi)大量 CPU 周期。推薦的方法是使用條件變量，即 pthread_cond_t 類型的變量。
復(fù)制代碼 代碼如下:
template typename T>
class BlockingQueue
{
public:
BlockingQueue ( )
{
pthread_mutexattr_init(_attr);
// set lock recursive
pthread_mutexattr_settype(_attr,PTHREAD_MUTEX_RECURSIVE_NP);
pthread_mutex_init(_lock,_attr);
pthread_cond_init(_cond, NULL);
}
~BlockingQueue ( )
{
pthread_mutex_destroy(_lock);
pthread_cond_destroy(_cond);
}
void push(const T data);
bool push(const T data, const int seconds); //time-out push
T pop( );
T pop(const int seconds); // time-out pop
private:
listT> _list;
pthread_mutex_t _lock;
pthread_mutexattr_t _attr;
pthread_cond_t _cond;
};
template typename T>
T BlockingQueueT>::pop( )
{
pthread_mutex_lock(_lock);
while (_list.empty( ))
{
pthread_cond_wait(_cond, _lock) ;
}
T _temp = _list.front( );
_list.pop_front( );
pthread_mutex_unlock(_lock);
return _temp;
}
template typename T>
void BlockingQueue T>::push(const T value )
{
pthread_mutex_lock(_lock);
const bool was_empty = _list.empty( );
_list.push_back(value);
pthread_mutex_unlock(_lock);
if (was_empty)
pthread_cond_broadcast(_cond);
}
并發(fā)阻塞隊(duì)列設(shè)計(jì)有兩個(gè)要注意的方面：
1.可以不使用 pthread_cond_broadcast，而是使用 pthread_cond_signal。但是，pthread_cond_signal 會釋放至少一個(gè)等待條件變量的線程，這個(gè)線程不一定是等待時(shí)間最長的讀線程。盡管使用 pthread_cond_signal 不會損害阻塞隊(duì)列的功能，但是這可能會導(dǎo)致某些讀線程的等待時(shí)間過長。
2.可能會出現(xiàn)虛假的線程喚醒。因此，在喚醒讀線程之后，要確認(rèn)列表非空，然后再繼續(xù)處理。強(qiáng)烈建議使用基于 while 循環(huán)的 pop()。
設(shè)計(jì)有超時(shí)限制的并發(fā)阻塞隊(duì)列
在許多系統(tǒng)中，如果無法在特定的時(shí)間段內(nèi)處理新數(shù)據(jù)，就根本不處理數(shù)據(jù)了。例如，新聞?lì)l道的自動收報(bào)機(jī)顯示來自金融交易所的實(shí)時(shí)股票行情，它每 n 秒收到一次新數(shù)據(jù)。如果在 n 秒內(nèi)無法處理以前的一些數(shù)據(jù)，就應(yīng)該丟棄這些數(shù)據(jù)并顯示最新的信息。根據(jù)這個(gè)概念，我們來看看如何給并發(fā)隊(duì)列的添加和取出操作增加超時(shí)限制。這意味著，如果系統(tǒng)無法在指定的時(shí)間限制內(nèi)執(zhí)行添加和取出操作，就應(yīng)該根本不執(zhí)行操作。
復(fù)制代碼 代碼如下:
template typename T>
bool BlockingQueue T>::push(const T data, const int seconds)
{
struct timespec ts1, ts2;
const bool was_empty = _list.empty( );
clock_gettime(CLOCK_REALTIME, ts1);
pthread_mutex_lock(_lock);
clock_gettime(CLOCK_REALTIME, ts2);
if ((ts2.tv_sec – ts1.tv_sec) seconds)
{
was_empty = _list.empty( );
_list.push_back(value);
}
pthread_mutex_unlock(_lock);
if (was_empty)
pthread_cond_broadcast(_cond);
}
template typename T>
T BlockingQueue T>::pop(const int seconds)
{
struct timespec ts1, ts2;
clock_gettime(CLOCK_REALTIME, ts1);
pthread_mutex_lock(_lock);
clock_gettime(CLOCK_REALTIME, ts2);
// First Check: if time out when get the _lock
if ((ts1.tv_sec – ts2.tv_sec) seconds)
{
ts2.tv_sec += seconds; // specify wake up time
while(_list.empty( ) (result == 0))
{
result = pthread_cond_timedwait(_cond, _lock, ts2) ;
}
if (result == 0) // Second Check: if time out when timedwait
{
T _temp = _list.front( );
_list.pop_front( );
pthread_mutex_unlock(_lock);
return _temp;
}
}
pthread_mutex_unlock(lock);
throw "timeout happened";
}
設(shè)計(jì)有大小限制的并發(fā)阻塞隊(duì)列
最后，討論有大小限制的并發(fā)阻塞隊(duì)列。這種隊(duì)列與并發(fā)阻塞隊(duì)列相似，但是對隊(duì)列的大小有限制。在許多內(nèi)存有限的嵌入式系統(tǒng)中，確實(shí)需要有大小限制的隊(duì)列。
對于阻塞隊(duì)列，只有讀線程需要在隊(duì)列中沒有數(shù)據(jù)時(shí)等待。對于有大小限制的阻塞隊(duì)列，如果隊(duì)列滿了，寫線程也需要等待。
復(fù)制代碼 代碼如下:
template typename T>
class BoundedBlockingQueue
{
public:
BoundedBlockingQueue (int size) : maxSize(size)
{
pthread_mutex_init(_lock, NULL);
pthread_cond_init(_rcond, NULL);
pthread_cond_init(_wcond, NULL);
_array.reserve(maxSize);
}
~BoundedBlockingQueue ( )
{
pthread_mutex_destroy(_lock);
pthread_cond_destroy(_rcond);
pthread_cond_destroy(_wcond);
}
void push(const T data);
T pop( );
private:
vectorT> _array; // or T* _array if you so prefer
int maxSize;
pthread_mutex_t _lock;
pthread_cond_t _rcond, _wcond;
};
template typename T>
void BoundedBlockingQueue T>::push(const T value )
{
pthread_mutex_lock(_lock);
const bool was_empty = _array.empty( );
while (_array.size( ) == maxSize)
{
pthread_cond_wait(_wcond, _lock);
}
_array.push_back(value);
pthread_mutex_unlock(_lock);
if (was_empty)
pthread_cond_broadcast(_rcond);
}
template typename T>
T BoundedBlockingQueueT>::pop( )
{
pthread_mutex_lock(_lock);
const bool was_full = (_array.size( ) == maxSize);
while(_array.empty( ))
{
pthread_cond_wait(_rcond, _lock) ;
}
T _temp = _array.front( );
_array.erase( _array.begin( ));
pthread_mutex_unlock(_lock);
if (was_full)
pthread_cond_broadcast(_wcond);
return _temp;
}
要注意的第一點(diǎn)是，這個(gè)阻塞隊(duì)列有兩個(gè)條件變量而不是一個(gè)。如果隊(duì)列滿了，寫線程等待 _wcond 條件變量；讀線程在從隊(duì)列中取出數(shù)據(jù)之后需要通知所有線程。同樣，如果隊(duì)列是空的，讀線程等待 _rcond 變量，寫線程在把數(shù)據(jù)插入隊(duì)列中之后向所有線程發(fā)送廣播消息。如果在發(fā)送廣播通知時(shí)沒有線程在等待 _wcond 或 _rcond，會發(fā)生什么？什么也不會發(fā)生；系統(tǒng)會忽略這些消息。還要注意，兩個(gè)條件變量使用相同的互斥鎖。
您可能感興趣的文章:- linux中高并發(fā)socket最大連接數(shù)的優(yōu)化詳解
- Linux netstat命令查看并發(fā)連接數(shù)的方法
- Linux下高并發(fā)socket最大連接數(shù)所受的各種限制(詳解)
- linux并發(fā)連接50萬的配置方法
- 淺談Linux環(huán)境下并發(fā)編程中C語言fork()函數(shù)的使用
- Linux下apache如何限制并發(fā)連接和下載速度
- Linux并發(fā)執(zhí)行很簡單,這么做就對了
