內(nèi)存消息隊(duì)列是服務(wù)器端常用的基礎(chǔ)組件，他使得符合生產(chǎn)者-消費(fèi)者模型的兩個(gè)線程或兩組線程之間的通訊看起來(lái)更加清晰，即生產(chǎn)者將消息壓入隊(duì)列，消費(fèi)者從隊(duì)列里面取走消息并處理，具體到網(wǎng)絡(luò)服務(wù)器結(jié)構(gòu)中，生產(chǎn)者線程是網(wǎng)絡(luò)接收線程，消費(fèi)者線程是邏輯處理線程，網(wǎng)絡(luò)線程不停的將接收到的數(shù)據(jù)放到全局消息隊(duì)列中，邏輯處理線程不停的從全局消息隊(duì)列中取走消息并處理。
system v消息隊(duì)列接口非常簡(jiǎn)單，主要是msgsnd,msgrcv，每個(gè)消息的結(jié)構(gòu)中都包含一個(gè)類型信息，這樣在msgrcv時(shí)就可以選擇只讀取某個(gè)類型的消息，如果類型傳為0，則不考慮類型，讀取第一個(gè)消息。根據(jù)消息類型來(lái)獲取消息是非常有用的，考慮下述簡(jiǎn)單服務(wù)器結(jié)構(gòu):
client server dbproxy
其中，server是主要的業(yè)務(wù)處理服務(wù)器，dbproxy是數(shù)據(jù)庫(kù)代理。以server為例，他需要處理兩方面的消息：1。來(lái)自client的消息；2。來(lái)自dbproxy的消息。定義如下枚舉：
enum eQueueType
{
QUEUE_TYPE_CLIENT = 1,
QUEUE_TYPE_WORLD = 2,
};
當(dāng)網(wǎng)絡(luò)線程收到來(lái)自client的消息時(shí)，將消息放到QUEUE_TYPE_CLIENT 類型的隊(duì)列中，當(dāng)收到dbproxy的消息時(shí)，放到QUEUE_TYPE_WORLD中，然后設(shè)置兩個(gè)或兩組線程分別處理
QUEUE_TYPE_CLIENT 隊(duì)列和QUEUE_TYPE_WORLD隊(duì)列，結(jié)構(gòu)非常清晰。在具體實(shí)現(xiàn)MessageQueue時(shí)，是基于ACE的消息隊(duì)列的，在內(nèi)部設(shè)置了一個(gè)消息隊(duì)列的map，也就是多個(gè)消息隊(duì)列，但在接口上就是一個(gè)隊(duì)列，MessageQueue只提供了根據(jù)類型來(lái)獲取隊(duì)列中的消息，沒(méi)有提供獲取整個(gè)隊(duì)列組中第一個(gè)消息的功能。
考慮使用全局消息隊(duì)列是我思考了一段時(shí)間才決定的，之前的做法是類似Active Object的，在ACE里就是ACE_Task類，該類即具備線程的功能，也包含了一個(gè)消息隊(duì)列，在使用時(shí)重載他的svc成員即可。對(duì)不熟悉ACE的人來(lái)說(shuō)，ACE_Task在設(shè)計(jì)上比較復(fù)雜，還牽扯到隊(duì)列的操作，如此以來(lái)，提供更友好的接口就顯得很有必要，因?yàn)榫€程的使用大家都已經(jīng)非常熟悉，消息隊(duì)列在概念上也很清晰，那么思路就是：我可以自由的創(chuàng)建線程，在線程里從全局消息隊(duì)列中讀取消息。看似和ACE_Task沒(méi)有根本區(qū)別，其實(shí)全局消息隊(duì)列的C風(fēng)格的接口，而不是派生的方式，大大降低了復(fù)雜度。
代碼貼上，方便以后查閱：
MessageQueue.h
#pragma once
#include <map>
#include <ace/Synch.h>
#include <ace/Message_Queue.h>
#include <ace/Singleton.h>
using std::map;
class MessageQueue
{
public:
MessageQueue(void);
~MessageQueue(void);
public:
static MessageQueue* Instance();
ACE_Message_Block* GetMessage(int nType);
ACE_Message_Block* GetMessage(int nType,int nSeconds );//超時(shí)nSeconds秒
bool PutMessage(int nType,ACE_Message_Block* pMsg);
bool PutMessage(int nType,ACE_Message_Block* pMsg,int nSeconds);//超時(shí)nSeconds秒
protected:
//獲取某類型的消息隊(duì)列，沒(méi)有則創(chuàng)建
ACE_Message_Queue<ACE_MT_SYNCH>* GetQueue(int nType);
private:
map<int,ACE_Message_Queue<ACE_MT_SYNCH>* > m_QueueMap;
ACE_Thread_Mutex m_QueueMapMutex;
};
typedef ACE_Singleton<MessageQueue,ACE_Thread_Mutex> MessageQueueSingleton;MessageQueue.cpp
#include <cassert>
#include <ace/Guard_T.h>
#include "MessageQueue.h"
// MessageQueue.cpp : 定義控制臺(tái)應(yīng)用程序的入口點(diǎn)。
//
MessageQueue::MessageQueue(void)
{
}
MessageQueue::~MessageQueue(void)
{
ACE_GUARD(ACE_Thread_Mutex,g,m_QueueMapMutex);
for (map<int,ACE_Message_Queue<ACE_MT_SYNCH>* >::iterator iter = m_QueueMap.begin(); iter != m_QueueMap.end();++iter)
{
ACE_Message_Queue<ACE_MT_SYNCH>* pQueue = iter->second;
pQueue->close();
delete pQueue;
}
m_QueueMap.clear();
}
MessageQueue* MessageQueue::Instance()
{
return MessageQueueSingleton::instance();
}
ACE_Message_Block* MessageQueue::GetMessage( int nType )
{
ACE_Message_Queue<ACE_MT_SYNCH>* pQueue = GetQueue(nType);
ACE_Message_Block* pMsg = NULL;
int nRet = pQueue->dequeue(pMsg);
if (nRet != -1)
{
return pMsg;
}
else
{
return NULL;
}
}
ACE_Message_Block* MessageQueue::GetMessage( int nType,int nSeconds )
{
assert( nSeconds > 0);
ACE_Time_Value timeout = ACE_OS::gettimeofday();
timeout += nSeconds;
ACE_Message_Queue<ACE_MT_SYNCH>* pQueue = GetQueue(nType);
ACE_Message_Block* pMsg = NULL;
int nRet = pQueue->dequeue(pMsg,&timeout);
if (nRet != -1)
{
return pMsg;
}
else
{
return NULL;
}
}
bool MessageQueue::PutMessage( int nType,ACE_Message_Block* pMsg )
{
ACE_Message_Queue<ACE_MT_SYNCH>* pQueue = GetQueue(nType);
int nRet = pQueue->enqueue(pMsg);
return nRet != -1;
}
bool MessageQueue::PutMessage( int nType,ACE_Message_Block* pMsg,int nSeconds )
{
assert( nSeconds > 0);
ACE_Time_Value timeout = ACE_OS::gettimeofday();
timeout += nSeconds;
ACE_Message_Queue<ACE_MT_SYNCH>* pQueue = GetQueue(nType);
int nRet = pQueue->enqueue(pMsg,&timeout);
return nRet != -1;
}
ACE_Message_Queue<ACE_MT_SYNCH>* MessageQueue::GetQueue( int nType )
{
assert( nType >= 0 );
ACE_GUARD_RETURN(ACE_Thread_Mutex,g,m_QueueMapMutex,NULL);
if (m_QueueMap.find(nType) != m_QueueMap.end())
{
return m_QueueMap[nType];
}
else
{
ACE_Message_Queue<ACE_MT_SYNCH>* pQueue = new ACE_Message_Queue<ACE_MT_SYNCH>();
m_QueueMap[nType] = pQueue;
return pQueue;
}
}test.cpp
#include "MessageQueue.h"
#include <iostream>
#include <ace/OS.h>
using namespace std;
enum eQueueType
{
QUEUE_TYPE_CLIENT = 1,
QUEUE_TYPE_WORLD = 2,
};
int main(int argc, char* argv[])
{
MessageQueue* pQueue = MessageQueue::Instance();
ACE_Message_Block* pClientMsg = new ACE_Message_Block(100);
pClientMsg->copy("client msg");
pQueue->PutMessage(QUEUE_TYPE_CLIENT,pClientMsg);
ACE_Message_Block* pWorldMsg = new ACE_Message_Block(100);
pWorldMsg->copy("world msg");
pQueue->PutMessage(QUEUE_TYPE_WORLD,pWorldMsg);
ACE_Message_Block* pTemp = NULL;
pTemp = pQueue->GetMessage(QUEUE_TYPE_CLIENT);
cout << pTemp->rd_ptr() << endl;
pTemp = pQueue->GetMessage(QUEUE_TYPE_WORLD);
cout << pTemp->rd_ptr() << endl;
cout << "begin time : " << ACE_OS::time(NULL) << endl;
pTemp = pQueue->GetMessage(QUEUE_TYPE_CLIENT,10);
cout << "end time : " << ACE_OS::time(NULL) << endl;
if (pTemp == NULL)
{
cout << "time out when get client msg" << endl;
}
return 0;
}