本例示范Linux信號量的基本用法。該范例使用了兩個(gè)線程分別對一個(gè)公用隊(duì)列進(jìn)行入隊(duì)和出隊(duì)操作，并用信號量進(jìn)行控制，當(dāng)隊(duì)列空時(shí)出隊(duì)操作可以被阻塞，當(dāng)隊(duì)列滿時(shí)入隊(duì)操作可以被阻塞。
主要用到的信號量函數(shù)有：
sem_init：初始化信號量sem_t，初始化的時(shí)候可以指定信號量的初始值，以及是否可以在多進(jìn)程間共享。
sem_wait：一直阻塞等待直到信號量>0。
sem_timedwait：阻塞等待若干時(shí)間直到信號量>0。
sem_post：使信號量加1。
sem_destroy：釋放信號量。和sem_init對應(yīng)。
關(guān)于各函數(shù)的具體參數(shù)請用man查看。如man sem_init可查看該函數(shù)的幫助。
下面看具體的代碼：
//--------------------------msgdequeue.h開始-------------------------------------
//實(shí)現(xiàn)可控隊(duì)列
#ifndef MSGDEQUEUE_H
#define MSGDEQUEUE_H
#include "tmutex.h"
#include <iostream>
#include <errno.h>
#include <time.h>
#include <semaphore.h>
#include <deque>
using namespace std;
template<typename T,typename MUTEX_TYPE = ThreadMutex>
class CMessageDequeue
{
public:
CMessageDequeue(size_t MaxSize) : m_MaxSize( MaxSize )
{
sem_init( &m_enques,0, m_MaxSize ); //入隊(duì)信號量初始化為MaxSize，最多可容納MaxSize各元素
sem_init( &m_deques,0,0 ); //隊(duì)列剛開始為空，出隊(duì)信號量初始為0
}
~CMessageDequeue()
{
sem_destroy(&m_enques);
sem_destroy(&m_deques);
}
int sem_wait_i( sem_t *psem, int mswait )
{//等待信號量變成>0，mswait為等待時(shí)間，若mswait<0則無窮等待，否則等待若干mswait毫秒。
if( mswait < 0 )
{
int rv = 0;
while( ((rv = sem_wait(psem) ) != 0 ) && (errno == EINTR
) ); //等待信號量，errno==EINTR屏蔽其他信號事件引起的等待中斷
return rv;
}
else
{
timespec ts;
clock_gettime(CLOCK_REALTIME, &ts ); //獲取當(dāng)前時(shí)間
ts.tv_sec += (mswait / 1000 ); //加上等待時(shí)間的秒數(shù)
ts.tv_nsec += ( mswait % 1000 ) * 1000; //加上等待時(shí)間納秒數(shù)
int rv = 0;
while( ((rv=sem_timedwait( psem, &ts ))!=0) && (errno ==
EINTR) ); //等待信號量，errno==EINTR屏蔽其他信號事件引起的等待中斷
return rv;
}
}
bool push_back( const T &item, int mswait = -1 )
{ //等待mswait毫秒直到將item插入隊(duì)列，mswait為-1則一直等待
if( -1 == sem_wait_i( &m_enques, mswait ))
{
return false;
}
//AUTO_GUARD：定界加鎖，見Linux多線程及臨界區(qū)編程例解的tmutex.h文件定義。
AUTO_GUARD( g, MUTEX_TYPE, m_lock );
try
{
m_data.push_back( item );
cout << "push " << item << endl;
sem_post( &m_deques );
return true;
}
catch(...)
{
return false;
}
}
bool pop_front( T &item, bool bpop = true, int mswait = -1 )
{ //等待mswait毫秒直到從隊(duì)列取出元素，mswait為-1則一直等待
if( -1 == sem_wait_i( &m_deques, mswait ) )
{
return false;
}
//AUTO_GUARD：定界加鎖，見Linux多線程及臨界區(qū)編程例解的tmutex.h文件定義。
AUTO_GUARD( g, MUTEX_TYPE, m_lock );
try
{
item = m_data.front();
if( bpop )
{
m_data.pop_front();
cout << "pop " << item << endl;
}
sem_post( &m_enques );
return true;
}
catch(...)
{
return false;
}
}
inline size_t size()
{
return m_data.size();
}
private:
MUTEX_TYPE m_lock;
deque<T> m_data;
size_t m_MaxSize;
sem_t m_enques;
sem_t m_deques;
};
#endif
//--------------------------msgdequeue.h結(jié)束-------------------------------------
//--------------------------test.cpp開始-------------------------------------
//主程序文件
#include "msgdequeue.h"
#include <pthread.h>
#include <iostream>
using namespace std;
CMessageDequeue<int> qq(5);
void *get_thread(void *parg);
void *put_thread(void *parg);
void *get_thread(void *parg)
{
while(true)
{
int a = -1;
if( !qq.pop_front( a,true, 1000 ) )
{
cout << "pop failed. size=" << qq.size() << endl;
}
}
return NULL;
}
void *put_thread(void *parg)
{
for(int i=1; i<=30; i++)
{
qq.push_back( i, -1 );
}
return NULL;
}
int main()
{
pthread_t pget,pput;
pthread_create( &pget,NULL,get_thread,NULL);
pthread_create( &pput, NULL, put_thread,NULL);
pthread_join( pget,NULL );
pthread_join( pput,NULL );
return 0;
}
//--------------------------test.cpp結(jié)束------------------------------------- 編譯程序：g++ msgdequeue.h test.cpp -lpthread -lrt -o test
-lpthread鏈接pthread庫。-ltr鏈接clock_gettime函數(shù)相關(guān)庫。
編譯后生成可執(zhí)行文件test。輸入./test執(zhí)行程序。
線程get_thread每隔1000毫秒從隊(duì)列取元素，線程put_thread將30個(gè)元素依次入隊(duì)。兩個(gè)線程模擬兩條入隊(duì)和出隊(duì)的流水線。因我們在 CMessageDequeue<int> qq(5)處定義了隊(duì)列最多可容納5個(gè)元素，入隊(duì)線程每入隊(duì)到隊(duì)列元素滿5個(gè)后需阻塞等待出隊(duì)線程將隊(duì)列元素出隊(duì)才能繼續(xù)。測試時(shí)可調(diào)整隊(duì)列可容納最大元素個(gè)數(shù)來觀察運(yùn)行效果。
posted on 2011-09-22 10:06
老馬驛站 閱讀(1018)
評論(0) 編輯 收藏 引用 所屬分類:
linux