关于RMsgQueue类的使用
RMsgQueue类是Symbian OS EKA2才提供的一个类，最近因为项目中要使用，为此对使用进行如下小结。
因为RMsgQueue类只是一个封装好的内核资源类，类似于RSocket和RTimer类，要想使用它进行异步操作就必须对其用AO类来封装，从而来实现监听消息，在有消息过来时得到通知并根据消息内容进行相对应的处理。
那这个消息内容又该如何定义呢？我们可以参看下RMsgQueue类定义
template <typename T>
class RMsgQueue : public RMsgQueueBase
{
public:
TInt CreateLocal(TInt aSize, TOwnerType aType=EOwnerProcess);
TInt CreateGlobal(const TDesC& aName, TInt aSize, TOwnerType aType=EOwnerProcess);
TInt Send(const T& aMsg);
void SendBlocking(const T& aMsg);
TInt Receive(T& aMsg);
void ReceiveBlocking(T& aMsg);
};
显然该类RMsgQueue也是Symbian的廋模板类，具体消息内容可以根据需求自定义类型，即如果传递的消息足够简单则可以采用Tint类型来定义，考虑到我们的实际应用对消息内容进行如下定义
typedef enum _NetWork_Msg_Type
{
NetworkSocketCreate,
NetworkSocketConnect,
NetworkSocketSend,
NetworkSocketRecv,
NetworkSocketClose,
}NetWorkMsgType;
typedef struct _NetWork_Msg
{
NetWorkMsgType m_msgType;
TInt32 m_msgFd;
}NetworkMsg;
其中的NetworkMsg就是我们在这里使用的消息内容，其中的成员m_msgType 是NetWorkMsgType定义的枚举值，表示消息类型，另一个成员m_msgFd是我们跨线程传递一个Id值。
有了以上消息内容，我们就可以简单的来定义消息队列了
RMsgQueue<NetworkMsg> g_NetWorkMsgQue;
如上已经实例化了一个RMsgQueue对象，但是RMsgQueue对象必须要像RTimer一样需要调用CreateLocal或者CreateGlobal才能创建起来，那我们在哪里创建它呢？这两个创建又有什么区别呢？由于RMsgQueue可以是全局的也可以是局部的，假如使用局部的，那么类的封装性较好，而且可以很好的跨进程使用，但是缺点是需要传递对象，此时创建就可以使用CreateGlobal。考虑到我们只在一个进程的不同线程间使用，同一进程的线程间可以共享资源（内存），所以这里我们就采用简单的全局变量来实现它，为此创建就用了简单的CreateLocal函数，而且我们将其放在了AO的二阶段构造中，详见示例代码。
至于AO的其它封装，主要是利用RMsgQueue如下三个函数
void RMsgQueue::NotifyDataAvailable(TRequestStatus& aStatus);
TInt RMsgQueue::Send(const T& aMsg);
TInt RMsgQueue::Receive(T& aMsg);
先在封装的AO中调用NotifyDataAvailable开启消息的监听，然后就开始等待外部的Send函数调用发消息进来，一旦有消息Send进来就进入AO的Runl中，我们通过调用Receive函数来对传递进来的消息内容进行解析和相应处理。逻辑就这么简单，下面给出源代码供参考。
头文件内容：
/*
* MessageQueueAO.h
*
* Created on: 2010-3-30
* Author: frank
*/
#ifndef MESSAGEQUEUEAO_H_
#define MESSAGEQUEUEAO_H_
#include <e32base.h>
#include <e32msgqueue.h>
typedef enum _NetWork_Msg_Type
{
NetworkConnect,
NetworkSocketCreate,
NetworkSocketConnect,
NetworkSocketSend,
NetworkSocketRecv,
NetworkSocketClose,
NetWorkDisConnect
}NetWorkMsgType;
typedef struct _NetWork_Msg
{
NetWorkMsgType m_msgType;
TInt32 m_msgFd;
}NetworkMsg;
const TInt KNumberOfMsgs = 10;
extern RMsgQueue<NetworkMsg> g_NetWorkMsgQue;
class CMessageQueueAO : public CActive
{
public:
// Cancel and destroy
virtual ~CMessageQueueAO();
// Two-phased constructor.
static CMessageQueueAO* NewL();
// Two-phased constructor.
static CMessageQueueAO* NewLC();
public:
// New functions
// Function for making the initial request
TInt StartMessageGet(const TDesC& aText=KNullDesC);
private:
// C++ constructor
CMessageQueueAO();
// Second-phase constructor
void ConstructL();
private:
// From CActive
// Handle completion
void RunL();
// How to cancel me
void DoCancel();
// Override to handle leaves from RunL(). Default implementation causes
// the active scheduler to panic.
TInt RunError(TInt aError);
};
#endif /* MESSAGEQUEUEAO_H_ */
实现文件内容
/*
* MessageQueueAO.cpp
*
* Created on: 2010-3-30
* Author: frank
*/
#include "MessageQueueAO.h"
RMsgQueue<NetworkMsg> g_NetWorkMsgQue;
CMessageQueueAO::CMessageQueueAO()
:CActive(EPriorityHigh)
{
// TODO Auto-generated constructor stub
}
CMessageQueueAO::~CMessageQueueAO()
{
// TODO Auto-generated destructor stub
Cancel();
}
CMessageQueueAO* CMessageQueueAO::NewL()
{
CMessageQueueAO* self = CMessageQueueAO::NewLC();
CleanupStack::Pop(); // self;
return self;
}
CMessageQueueAO* CMessageQueueAO::NewLC()
{
CMessageQueueAO* self = new (ELeave) CMessageQueueAO();
CleanupStack::PushL(self);
self->ConstructL();
return self;
}
void CMessageQueueAO::ConstructL()
{
CActiveScheduler::Add(this); // Add to scheduler
g_NetWorkMsgQue.CreateLocal(KNumberOfMsgs);
}
TInt CMessageQueueAO::StartMessageGet(const TDesC& aText)
{
g_NetWorkMsgQue.NotifyDataAvailable(iStatus);
SetActive(); // Tell scheduler a request is active
}
void SocketCreate(TInt32 aFd)
{
}
void SocketRecv(TInt32 aFd)
{
}
void SocketConnect(TInt32 aFd)
{
}
void CMessageQueueAO::RunL()
{
if (iStatus.Int() == KErrNone)
{
//接收数据
NetworkMsg msgfrmq;
g_NetWorkMsgQue.Receive(msgfrmq);
switch(msgfrmq.m_msgType)
{
case NetworkSocketCreate:
LockMutexRaw(g_NetMgrMutex);
SocketCreate(msgfrmq.m_msgFd);
UnlockMutexRaw(g_NetMgrMutex);
break;
case NetworkSocketConnect:
LockMutexRaw(g_NetMgrMutex);
SocketConnect(msgfrmq.m_msgFd);
UnlockMutexRaw(g_NetMgrMutex);
break;
case NetworkSocketRecv:
LockMutexRaw(g_NetMgrMutex);
SocketRecv(msgfrmq.m_msgFd);
UnlockMutexRaw(g_NetMgrMutex);
break;
case NetworkSocketClose:
LockMutexRaw(g_NetMgrMutex);
SocketClose(msgfrmq.m_msgFd);
UnlockMutexRaw(g_NetMgrMutex);
break;
default:
break;
}
//Using ReceiveBlocking()
// msgqueue.ReceiveBlocking(msgfrmq);// 如果使用blocking 可能那边要发送两次消息这边才会跑哦
}
g_NetWorkMsgQue.NotifyDataAvailable(iStatus);//开启下一个消息的接收
SetActive();
}
void CMessageQueueAO::DoCancel()
{
if( IsActive() )
{
g_NetWorkMsgQue.CancelDataAvailable();
}
}
TInt CMessageQueueAO::RunError(TInt aError)
{
return aError;
}
封装完这个AO，大功就可以完成了，将这个AO放在一个独立的线程中创建起来，并调用StartMessageGet让这个线程始终监听消息。外界线程通过调用g_NetWorkMsgQue.Send(msgfrmq)来给这个AO线程发消息并得到相应的处理。在这里我就不过多展开了，因为要涉及两个线程间的操作，篇幅就太大了。
在结束小结的时候，我想着如何实现IPC，这样的话，这个小结就完整了，碰巧，在逛博客的时候，发现一篇博文总结得很好，而且言简意赅，为此转载与下面，跟大家一起分享，博文原址http://blog.sina.com.cn/s/blog_63b4ee0d0100g3xc.html
如何理解进程间通信？
先用一个通俗的例子来解释：比如我们需要实现这样一种模式，应用由进程A与进程B两部分组成，A有UI，负责用户交互；B没有UI，完全后台运行，A与B之间可以相互通信。就如彩信的模式，彩信到达后后台下载，下载完给出提示信息（即用户界面），阅读彩信再激活“信息”程序。
进程通信即两个并行进程可以通过互相发送消息进行合作，消息是通过消息缓冲而在进程之间相互传递的。
RMsgQueue是3版提供的比较好的与事件机制融合的技术, 2版只能使用其它传统的, 信号量, 共享内存等技术.
解决方案：
让我们来看代码:
Server
void CP2PServer::ConstructL()
{
//创建
iMsgQueue.CreateGlobal(KGLobalName, KNumberOfSlots, KMessageLength, EOwnerProcess);
CActiveScheduler::Add( this); // Add to scheduler
iMsgQueue.NotifyDataAvailable( iStatus );//开始监听消息
SetActive(); // Tell scheduler a request is active
}
void CP2PServer::RunL()
{
if (iStatus.Int() == KErrNone)
{
//接收数据
TRAPD(error,iMsgQueue.Receive( &str_SendData, KMessageLength));
if(error==KErrNone)
{
iObserver->HandleMessageReceiveL(str_SendData.DataBuf.Left(str_SendData.DataLength));
}
}
}
Client
void CP2PClient::ConstructL()
{
//创建
iMsgQueue.OpenGlobal(KGLobalName, EOwnerProcess);
}
//发送消息
void CP2PClient::SendMessageL(const TDesC8& aSendMessage)
{
STR_SENDDATA temp;
temp.DataLength = aSendMessage.Length();
temp.DataBuf.Copy(aSendMessage);
iMsgQueue.Send(&temp, KMessageLength);
}
就一些疑难问题解决如下：
B线程创建一个“东东”,名字为KGLobalName，并且B线程开启监控消息， A线程在需要时使用带KGLobalName参数的OpenGlobal函数打开同名, 然后发送消息出来, 之后直接关闭，B会收到此消息解析后开始进行相应的处理。
上面的这个“东东”就是一个消息队列, 只要创建者线程B存在, 就可以使用, 一般这个设计都是B始终运行, 并且建立RMsgQueue, 其它进程需要时打开, 发送消息，然后关闭。
如果双方都可以作为发送方与接受方，则每方都创建RMsgQueue,注意使用不同的名字。
最后除了感谢上面这位牛人给出的总结外，我再补充一下，如果不是为了避免传递消息时的内存拷贝操作，单向操作用C/S架构更为可取，至于双向操作，目前看来的确还是RMsgQueue好些，不知道大牛们觉得我这种猜想是否成立。
posted on 2010-04-09 21:40
frank.sunny 阅读(2013)
评论(1) 编辑 收藏 引用 所属分类:
symbian 开发