franksunny的个人技术空间
获得人生中的成功需要的专注与坚持不懈多过天才与机会。 ——C.W. Wendte

 

关于RMsgQueue类的使用

 

RMsgQueue类是Symbian OS EKA2才提供的一个类,最近因为项目中要使用,为此对使用进行如下小结。

因为RMsgQueue类只是一个封装好的内核资源类,类似于RSocketRTimer类,要想使用它进行异步操作就必须对其用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进来就进入AORunl中,我们通过调用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两部分组成,AUI,负责用户交互;B没有UI,完全后台运行,AB之间可以相互通信。就如彩信的模式,彩信到达后后台下载,下载完给出提示信息(即用户界面),阅读彩信再激活信息程序。

进程通信即两个并行进程可以通过互相发送消息进行合作,消息是通过消息缓冲而在进程之间相互传递的。

RMsgQueue3版提供的比较好的与事件机制融合的技术, 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 阅读(1944) 评论(1)  编辑 收藏 引用 所属分类: symbian 开发

FeedBack:
# re: 关于RMsgQueue类的使用[未登录]
2011-09-14 10:19 | zzg
总结的不错,学习,转走了  回复  更多评论
  

只有注册用户登录后才能发表评论。
【推荐】超50万行VC++源码: 大型组态工控、电力仿真CAD与GIS源码库
网站导航: 博客园   IT新闻   BlogJava   知识库   博问   管理



常用链接

留言簿(13)

随笔分类

个人其它博客

基础知识链接

最新评论

阅读排行榜

评论排行榜