ACE_Task介绍（生产者/消费者）v3.0

ACE_Task面向对象的线程

ACE_Task的四个主要方法：

    int open(void*);//定制启动时分配资源和初始化，如果创建线程，在此方法里面调用 activate

    //virtual int activate (long flags,...);启动线程

    int svc(void);//线程运行就是执行此函数

    int close(u_long);//svc退出之后会自动调用此方法，长在此释放资源，线程数是多个的时候不可以直接在close里面delete this

参考：http://blog.csdn.net/calmreason/article/details/36399697

多线程的常用方法

等待所有线程退出
下面的这句话通常写在main的最后，表示当所有线程都执行结束的时候程序才退出（否则，某一个线程执行结束程序可能就退出了）。

ACE_Thread_Manager::instance()->wait();

退出当前线程：
下面的这句话写在线程执行的地方
（1）Task的svc方法里面
（2）函数里面，然后用这个函数创建线程。
会让当前线程直接退出。

ACE_Thread_Manager::instance()->exit();

与ACE_Task_Base的关系

ACE_Task_Base是主动对象的基类，ACE_Task继承了ACE_Task_Base的线程功能之后添加了具有同步策略功能的消息队列ACE_Message_Queue。如果你只需要一个线程对象，你可以直接使用ACE_Task_Base

同步策略

ACE_Task可以启动一个或多个线程，以及一个底层消息队列。各个任务通过消息队列进行通信。至于消息队列实现的内在细节程序员不必关注。

putq()

 将消息插入到另一任务的消息队列中

getq()

将消息提取出来

这样的体系结构大大简化了多线程程序的编程模型。

同步模式

分两种： ACE_MT_SYNCH（多线程）和ACE_NULL_SYNCH（单线程）。

多线程模式下线程的消息队列会使用多线程同步策略，会造成线程的阻塞；单线程模式下不存在同步的额外开销；多线程下保证一个线程对象在同一时刻只有一个方法在执行。

创建和使用Task的步骤

（1）编写一个派生自ACE_Task的类
（2）指定同步策略

（3）activate启动线程

此方法一般在open中调用，调用完activate方法之后svc方法自动运行。

指定线程的创建标志：在activate方法执行的时候可以指定线程的内部类型，THR_DETACHED（分离的，可以直接被ACE_Thread_Manager::wait()方法来回收），默认情况下线程的内部类型是THR_JOINABLE(可结合的，此线程退出的状态会被其他线程捕获并作出相应的处理)；THR_NEW_LWP （挂钩到内核级线程，会创建一个内核线程）；你的线程如果在独立运行一般你会使用：activate(THR_NEW_LWP | THR_BOUND | THR_DETACHED,1);来创建你的线程，1表示创建一个线程。

（4）svc运行线程

重载 ACE_Task的 svc 方法，编写消息循环相关的代码

（5）操作消息队列

ACE_Task<ACE_MT_SYNCH>   类自带一个消息队列
取消息的方法是：this->getq(blk);

放消息的方法是：this->putq(blk);//注意不是this->put(blk);

生产者消费者实例1：生产者和消费者共享同一个内部消息队列
生产者
ProduceAudio.h

#ifndef PRODUCEAUDIO_H
#define PRODUCEAUDIO_H

#include "ace/Task.h"

class ProduceAudio :  public ACE_Task<ACE_MT_SYNCH>
{
public:
ProduceAudio(ACE_Thread_Manager *thr_man=0,
ACE_Message_Queue<ACE_MT_SYNCH> *mq=0);
~ProduceAudio(void);
int open(void*);
int svc(void);
};

#endif

ProduceAudio.cpp

#include "ProduceAudio.h"

#include "ace/Log_Msg.h"
#include "ace/OS.h"
#include "Converter.h"
#include <string>
using namespace std;

ProduceAudio::ProduceAudio(ACE_Thread_Manager *thr_man,
ACE_Message_Queue<ACE_MT_SYNCH> *mq)
:ACE_Task<ACE_MT_SYNCH>(thr_man,mq)
{
}

ProduceAudio::~ProduceAudio(void)
{
ACE_DEBUG((LM_DEBUG, "(%t) ~ProduceAudio()\n"));

}

int ProduceAudio::open(void*)
{
ACE_DEBUG((LM_DEBUG, "(%t) ProduceAudio task opened\n"));
activate(THR_NEW_LWP,1);
return 0;
}

int ProduceAudio::svc(void)
{
ACE_DEBUG((LM_DEBUG, "(%t) ProduceAudio::svc() running\n"));
string s("message");
for ( int i=0;i<3;++i)
{
ACE_Message_Block * blk = new ACE_Message_Block(10);
blk->copy( (s + lexical_cast<string>(i)).c_str());
this->putq(blk);
//this->put(blk);
ACE_DEBUG((LM_DEBUG, "(%t) ProduceAudio::svc() put(%s),now msg_queue()->message_count()[%d]\n",blk->rd_ptr(),
this->msg_queue()->message_count()));
ACE_OS::sleep(1);
}
ACE_DEBUG((LM_DEBUG, "(%t) ProduceAudio::svc() return\n"));
return 0;
}

消费者
SendToServer.h

#ifndef SENDTOSERVER_H
#define SENDTOSERVER_H
#include "ace/Task.h"

class SendToServer :  public ACE_Task<ACE_MT_SYNCH>
{
public:
SendToServer(ACE_Thread_Manager *thr_man=0,
ACE_Message_Queue<ACE_MT_SYNCH> *mq=0);
~SendToServer(void);
int open(void*);
int svc(void);
};

#endif

SendToServer.cpp

#include "SendToServer.h"

#include "ace/OS.h"
#include <string>
using namespace std;

SendToServer::SendToServer(ACE_Thread_Manager *thr_man,
ACE_Message_Queue<ACE_MT_SYNCH> *mq)
:ACE_Task<ACE_MT_SYNCH>(thr_man,mq)
{
}

SendToServer::~SendToServer(void)
{
ACE_DEBUG((LM_DEBUG, "(%t) ~SendToServer()\n"));
}

int SendToServer::open(void*)
{
ACE_DEBUG((LM_DEBUG, "(%t) SendToServer task opened\n"));
activate(THR_NEW_LWP,1);
return 0;
}

int SendToServer::svc(void)
{
ACE_DEBUG((LM_DEBUG, "(%t) SendToServer::svc() running\n"));
ACE_Message_Block * blk = NULL;
int count =0;
for ( ; count<3;)
{
if (this->msg_queue()->message_count()>0)
{
this->getq(blk);
++count;
ACE_DEBUG((LM_DEBUG,"SendToServer get :%s\n",blk->rd_ptr()));
blk->release();
}
ACE_OS::sleep(1);
}
ACE_DEBUG((LM_DEBUG, "(%t) SendToServer::svc() return\n"));
return 0;
}

主函数main.cpp

#include "ace/Thread_Manager.h"
#include "SendToServer.h"
#include "ProduceAudio.h"

#ifdef _DEBUG
#pragma comment (lib,"ACEd.lib")
#else
#pragma comment (lib,"ACE.lib")
#endif

int main(int argc, char* argv[])
{
SendToServer consumer(NULL,NULL);
ProduceAudio producer(NULL,consumer.msg_queue());
producer.open(NULL);
consumer.open(NULL);

ACE_Thread_Manager::instance()->wait();
return 0;
}

生产者消费者实例2：生产者通过引用消费者，来操作消费者的内部消息队列

#ifdef _DEBUG
#pragma comment (lib,"ACEd.lib")
#else
#pragma comment (lib,"ACE.lib")
#endif

#include "ace/Log_Msg.h"
#include "ace/Task.h"
#include "ace/OS.h"
#include "ace/Message_Block.h"
#include <stdio.h>
#include <string.h>
#include <iostream>
#include <string>
#include <sstream>
using namespace std;

class My_Data
{
public:
My_Data(){key = ++id;cout<<"My_Data("<<id<<")\n";}
~My_Data(){cout<<"~My_Data("<<id<<")\n";}
string data;
int key;
static int id;
};
int My_Data::id = 0;

class Consumer:
public ACE_Task<ACE_MT_SYNCH>
{
public:
//启动Task消费线程
int open(void*)
{
ACE_DEBUG((LM_DEBUG, "(%t) Consumer task opened\n"));
activate(THR_NEW_LWP,1);
return 0;
}
int svc(void)
{
//Get ready to receive message from Producer
do
{
ACE_Message_Block * msg =0;
ACE_DEBUG((LM_DEBUG,"(%t)消费者开始取消息\n"));
if (!this->msg_queue()->is_empty())//取消息的时候最好要判断队列是否为空，因为如果刚开始取就是空的，就会阻塞，后来没有人唤醒的话就会一直阻塞
{
this->getq(msg);//从消息队列中取出一个消息，这个消息的内存使用权就转接到消息指针上面了。
ACE_DEBUG((LM_DEBUG,"(%t)消费者收到消息: 内容[%s]\n",msg->rd_ptr()));
msg->release();
}else
{
cout<<"队列空，等待10秒之后再取消息！"<<endl;
ACE_OS::sleep(10);
}

}while(true);
return 0;
}
int close(u_long)
{
ACE_DEBUG((LM_DEBUG,"Consumer closes down\n"));
return 0;
}
};
class Producer : public ACE_Task<ACE_MT_SYNCH>
{
public:
Producer(Consumer * consumer):consumer_(consumer){}
int open(void*)
{
ACE_DEBUG((LM_DEBUG, "(%t) Producer task opened\n"));
activate(THR_NEW_LWP,1);
return 0;
}
//The Service Processing routine
int svc(void)
{
//生产者深入一个用户名，放到消费者的队列中
do
{
My_Data one_data;
ACE_OS::sleep(1);//防止CPU使用率过高
ostringstream os;
os<<one_data.key;
one_data.data = "name" + os.str();

ACE_Message_Block* mb = new ACE_Message_Block(100);
mb->copy(one_data.data.c_str());
cout<<"将"<<mb->rd_ptr()<<"放入到了队列中\n";
this->consumer_->putq(mb);
} while (shutdown);
return 0;
}
int close(u_long)
{
ACE_DEBUG((LM_DEBUG,"Producer closes down\n"));
return 0;
}
private:
Consumer * consumer_;
};
int main(int argc, char * argv[])
{
Consumer * consumer = new Consumer;
Producer * producer = new Producer(consumer);
producer->open(0);
consumer->open(0);
//Wait for all the tasks to exit.
ACE_Thread_Manager::instance()->wait();
ACE_OS::system("pause");
delete producer;
delete consumer;
return 0;
}

分析：

以上为经典的生产者-消费者例子，演示了两个任务如何使用底层的消息队列进行通信。我们可以将生产者和消费者看作是不同的ACE_Task类型的对象。方案十分简单，但却是面向对象的，在编写面向对象的多线程程序或主动对象的实例时，我们可采用此方案，它提供了比低级线程API更好的方法。

多个Task共用一个消息队列

多个Task共用一个消息队列，可以使用ACE_Task的接口来方便的实现

SendToServer consumer(NULL,NULL);
ProduceAudio producer(NULL,consumer.msg_queue());
producer.msg_queue()->high_water_mark((size_t)(1024*1024*2));
consumer.open(NULL);
producer.open(NULL);

上面的consumer、producer都是ACE_Task的派生类对象，注意是生产者将自己的消息队列指定为消费者的，这样消费者消费自己的数据的时候其实就是生产者生产的了。

参考博客：

http://baike.baidu.com/view/9503964.htm

http://blog.csdn.net/imjj/article/details/1097248