設(shè)計(jì)并發(fā)隊(duì)列
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:
list<T> _list;
pthread_mutex_t _lock;
};
template <typename T>
void Queue<T>::push(const T& value )
{
pthread_mutex_lock(&_lock);
_list.push_back(value);
pthread_mutex_unlock(&_lock);
}
template <typename T>
T Queue<T>::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 類。
template <typename T>
void Queue<T>::push(const T& value )
{
pthread_mutex_lock(&_wlock);
_list.push_back(value);
pthread_mutex_unlock(&_wlock);
}
template <typename T>
T Queue<T>::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 類型的變量。
private:
list<T> _list;
pthread_mutex_t _lock;
pthread_mutexattr_t _attr;
pthread_cond_t _cond;
};
template <typename T>
T BlockingQueue<T>::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ù)了。例如，新聞頻道的自動收報(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í)行操作。