ACE程序员教程-第9章 消息队列(Message Queue)

#ifndef MQ_EG1_H_

#define MQ_EG1_H_

#include "ace/Message_Queue.h"

class QTest

{

public:

//Constructor creates a message queue with no synchronization

QTest(int num_msgs);

//Enqueue the num of messages required onto the message mq.

int enq_msgs();

//Dequeue all the messages previously enqueued.

int deq_msgs ();

private:

//Underlying message queue

ACE_Message_Queue *mq_;

//Number of messages to enqueue.

int no_msgs_;

};

#endif /*MQ_EG1.H_*/

#include "mq_eg1.h"

QTest::QTest(int num_msgs)

:no_msgs_(num_msgs)

{

ACE_TRACE("QTest::QTest");

//First create a message queue of default size.

if(!(this->mq_=new ACE_Message_Queue ()))

ACE_DEBUG((LM_ERROR,"Error in message queue initialization \n"));

}

{

ACE_TRACE("QTest::enq_msg");

for(int i=0; i

{

//create a new message block specifying exactly how large

//an underlying data block should be created.

ACE_Message_Block *mb;

ACE_NEW_RETURN(mb,
ACE_Message_Block(ACE_OS::strlen("This is message 1\n")),-1);　

//Insert data into the message block using the wr_ptr

ACE_OS::sprintf(mb->wr_ptr(), "This is message %d\n", i);

　

//Be careful to advance the wr_ptr by the necessary amount.

//Note that the argument is of type "size_t" that is mapped to

//bytes.

mb->wr_ptr(ACE_OS::strlen("This is message 1\n"));

　

//Enqueue the message block onto the message queue

if(this->mq_->enqueue_prio(mb)==-1)

{

ACE_DEBUG((LM_ERROR,"\nCould not enqueue on to mq!!\n"));

return -1;

}　

ACE_DEBUG((LM_INFO,"EQ’d data: %s\n", mb->rd_ptr() ));

} //end for

//Now dequeue all the messages

this->deq_msgs();

return 0;

}

{

ACE_TRACE("QTest::dequeue_all");

ACE_DEBUG((LM_INFO,"No. of Messages on Q:%d Bytes on Q:%d \n"

,mq_->message_count(),mq_->message_bytes()));

ACE_Message_Block *mb;

　

//dequeue the head of the message queue until no more messages are

//left. Note that I am overwriting the message block mb and I since

//I am using the dequeue_head() method I dont have to worry about

//resetting the rd_ptr() as I did for the wrt_ptr()

for(int i=0;i

{

mq_->dequeue_head(mb);

ACE_DEBUG((LM_INFO,"DQ’d data %s\n", mb->rd_ptr() ));

　

//Release the memory associated with the mb

mb->release();

}

return 0;

}

int main(int argc, char* argv[])

{

if(argc <2)

ACE_ERROR_RETURN((LM_ERROR, "Usage %s num_msgs", argv[0]), -1);

　

QTest test(ACE_OS::atoi(argv[1]));

if(test.enq_msgs() == -1)

ACE_ERROR_RETURN( (LM_ERROR,"Program failure \n"), -1);

}

　　分别有两个方法可用于设置和获取高低水位标：

//get the high water mark

size_t high_water_mark(void)

//set the high water mark

void high_water_mark(size_t hwm);

//get the low water_mark

size_t low_water_mark(void)

//set the low water_mark

void low_water_mark(size_t lwm)

　　下面的例子演示水位标和迭代器的使用：

#include ”ace/Get_Opt.h”

#include ”ace/Malloc_T.h”

#define SIZE_BLOCK 1

class Args

{

public:

Args(int argc, char*argv[],int& no_msgs, ACE_Message_Queue* &mq)

{

ACE_Get_Opt get_opts(argc,argv,”h:l:t:n:xsd”);

while((opt=get_opts())!=-1)

switch(opt)

{

case ’n’:

//set the number of messages we wish to enqueue and dequeue

no_msgs=ACE_OS::atoi(get_opts.optarg);

ACE_DEBUG((LM_INFO,”Number of Messages %d \n”,no_msgs));

break;

　

case ’h’:

//set the high water mark

hwm=ACE_OS::atoi(get_opts.optarg);

mq->high_water_mark(hwm);

ACE_DEBUG((LM_INFO,”High Water Mark %d msgs \n”,hwm));

break;

　

case ’l’:

//set the low water mark

lwm=ACE_OS::atoi(get_opts.optarg);

mq->low_water_mark(lwm);

ACE_DEBUG((LM_INFO,”Low Water Mark %d msgs \n”,lwm));

break;

　

default:

ACE_DEBUG((LM_ERROR,

”Usage -n -h -l\n”));

break;

}

}

private:

int opt;

int hwm;

int lwm;

};

class Qtest

{

public:

QTest(int argc, char*argv[])

{

//First create a message queue of default size.

if(!(this->mq_=new ACE_Message_Queue ()))

ACE_DEBUG((LM_ERROR,”Error in message queue initialization \n”));

　

//Use the arguments to set the water marks and the no of messages

args_ = new Args(argc,argv,no_msgs_,mq_);

}

int start_test()

{

for(int i=0; i

{

//Create a new message block of data buffer size 1

ACE_Message_Block * mb= new ACE_Message_Block(SIZE_BLOCK);

　

//Insert data into the message block using the rd_ptr

*mb->wr_ptr()=i;

　

//Be careful to advance the wr_ptr

mb->wr_ptr(1);

　

//Enqueue the message block onto the message queue

if(this->mq_->enqueue_prio(mb)==-1)

{

ACE_DEBUG((LM_ERROR,”\nCould not enqueue on to mq!!\n”));

return -1;

}

　

ACE_DEBUG((LM_INFO,”EQ’d data: %d\n”,*mb->rd_ptr()));

}

　

//Use the iterators to read

this->read_all();

　

//Dequeue all the messages

this->dequeue_all();

　

return 0;

}

void read_all()

{

ACE_DEBUG((LM_INFO,”No. of Messages on Q:%d Bytes on Q:%d \n”

,mq_->message_count(),mq_->message_bytes()));

ACE_Message_Block *mb;

　

//Use the forward iterator

ACE_DEBUG((LM_INFO,”\n\nBeginning Forward Read \n”));

ACE_Message_Queue_Iterator mq_iter_(*mq_);

while(mq_iter_.next(mb))

{

mq_iter_.advance();

ACE_DEBUG((LM_INFO,”Read data %d\n”,*mb->rd_ptr()));

}

　

//Use the reverse iterator

ACE_DEBUG((LM_INFO,”\n\nBeginning Reverse Read \n”));

ACE_Message_Queue_Reverse_Iterator

mq_rev_iter_(*mq_);

while(mq_rev_iter_.next(mb))

{

mq_rev_iter_.advance();

ACE_DEBUG((LM_INFO,”Read data %d\n”,*mb->rd_ptr()));

}

}

void dequeue_all()

{

ACE_DEBUG((LM_INFO,”\n\nBeginning DQ \n”));

ACE_DEBUG((LM_INFO,”No. of Messages on Q:%d Bytes on Q:%d \n”,

mq_->message_count(),mq_->message_bytes()));

ACE_Message_Block *mb;

　

//dequeue the head of the message queue until no more messages

//are left

for(int i=0;i

{

mq_->dequeue_head(mb);

ACE_DEBUG((LM_INFO,”DQ’d data %d\n”,*mb->rd_ptr()));

}

}

private:

Args *args_;

ACE_Message_Queue *mq_;

int no_msgs_;

};

int main(int argc, char* argv[])

{

QTest test(argc,argv);

if(test.start_test()<0)

ACE_DEBUG((LM_ERROR,”Program failure \n”));

}