一、概述
ACE_Task是ACE中的任务或主动对象“处理结构”的基类。ACE使用此类来实现主动对象模式。所有希望成为“主动对象”的对象都必须由此类派生。同时可将它看作是更高级的、更为面向对象的线程。
ACE_Task处理的是对象，因此更有利于构造OO程序，产生更好的OO软件，而且，它还包括了一种用于
与其他任务通信的易于使用的机制。
ACE_Task可用作：
<1>更高级的线程(常称其为任务)
<2>主动对象模式中的主动对象
PS.ACE任务：
每个任务都含有一或多个线程，以及一个底层消息队列。各个任务通过消息队列进行通信。至于消息队列实现的内在细节程序员不必关注。发送任务用putq() 将消息插入到另一任务的消息队列中，接收任务通过使用getq()将消息提取出来。这样的体系结构大大简化了多线程程序的编程模型。
二、创建和使用任务的步骤
从ACE_Task类派生的子类应实现以下业务逻辑：
<1>实现服务初始化和终止方法。
open()方法应该包含所有专属于任务的初始化代码。其中可能包括诸如连接控制块、锁和内存这样的资源。close()方法用于终止。
<2>调用启用(Activation)方法。
在主动对象实例化后，必须通过调用activate()启用它。要在主动对象中创建的线程数目及其它参数，被传递给activate()方法。它将使svc()方法成为所有它生成的线程的启动点。
<3>实现服务专有的处理方法。
在主动对象被启用后，各个新线程在 svc() 方法中启动。程序员必须在子类中定义此方法。
三、一个任务间通信的实例
//消费者类定义
＃i nclude "ace/OS.h"
＃i nclude "ace/Task.h"
＃i nclude "ace/Message_Block.h"
＃i nclude "ace/Task.h"
＃i nclude "ace/Message_Block.h"
//The Consumer Task.
class Consumer :
public ACE_Task<ACE_MT_SYNCH>
{
public:
int open(void*)
{
ACE_DEBUG((LM_DEBUG, "(%t) Producer task opened /n"));
class Consumer :
public ACE_Task<ACE_MT_SYNCH>
{
public:
int open(void*)
{
ACE_DEBUG((LM_DEBUG, "(%t) Producer task opened /n"));
//Activate the Task
activate(THR_NEW_LWP,1);
activate(THR_NEW_LWP,1);
return 0;
}
}
//The Service Processing routine
int svc(void)
{
//Get ready to receive message from Producer
ACE_Message_Block * mb = 0;
do
{
mb = 0;
int svc(void)
{
//Get ready to receive message from Producer
ACE_Message_Block * mb = 0;
do
{
mb = 0;
//Get message of underlying queue
getq(mb);
ACE_DEBUG((LM_DEBUG,
"(%t) Got message: %d from remote task/n", *mb->rd_ptr()));
}while(*mb->rd_ptr()<10);
getq(mb);
ACE_DEBUG((LM_DEBUG,
"(%t) Got message: %d from remote task/n", *mb->rd_ptr()));
}while(*mb->rd_ptr()<10);
return 0;
}
}
int close(u_long)
{
ACE_DEBUG((LM_DEBUG, "Consumer closes down /n"));
return 0;
}
};
{
ACE_DEBUG((LM_DEBUG, "Consumer closes down /n"));
return 0;
}
};
//生产者类定义
class Producer :
public ACE_Task<ACE_MT_SYNCH>
{
public:
Producer(Consumer * consumer) :
consumer_(consumer), data_(0)
{
mb = new ACE_Message_Block((char *)&data_, sizeof(data_));
}
public ACE_Task<ACE_MT_SYNCH>
{
public:
Producer(Consumer * consumer) :
consumer_(consumer), data_(0)
{
mb = new ACE_Message_Block((char *)&data_, sizeof(data_));
}
int open(void *)
{
ACE_DEBUG((LM_DEBUG, "(%t) Producer task opened /n"));
{
ACE_DEBUG((LM_DEBUG, "(%t) Producer task opened /n"));
//Activate the Task
activate(THR_NEW_LWP, 1);
activate(THR_NEW_LWP, 1);
return 0;
}
}
//The Service Processing routine
int svc(void)
{
while(data_ < 11)
{
//Send message to consumer
int svc(void)
{
while(data_ < 11)
{
//Send message to consumer
ACE_DEBUG((LM_DEBUG,
"(%t) Sending message: %d to remote task/n", data_));
consumer_->putq(mb_);
"(%t) Sending message: %d to remote task/n", data_));
consumer_->putq(mb_);
//Go to sleep for a sec.
ACE_OS::sleep(1);
data_++;
}
return 0;
}
}
int close(void)
{
ACE_DEBUG((LM_DEBUG, "Producer closes down /n"));
{
ACE_DEBUG((LM_DEBUG, "Producer closes down /n"));
return 0;
}
}
private:
char data_;
Consumer * consumer_;
ACE_Message_Block * mb_;
};
char data_;
Consumer * consumer_;
ACE_Message_Block * mb_;
};
//main()函数
int main(int argc, char *argv[])
{
Consumer * consumer = new Consumer;
{
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::exit(0);
}
}
分析：
以上为经典的生产者-消费者例子，演示了两个任务如何使用底层的消息队列进行通信。我们可以将生产者和消费者看作是不同的ACE_Task类型的对象。方案十分简单，但却是面向对象的，在编写面向对象的多线程程序或主动对象的实例时，我们可采用此方案，它提供了比低级线程API更好的方法。
ACE_Task封装了任务,每个任务都含有一或多个线程，以及一个底层消息队列。各个任务通过这些消息队列进行通信。
其主要成员如下:
open():初始化资源
close():释放资源
activate():启动线程,可指定线程的数目
svc():线程的启动位置
putq():放置消息到任务的消息队列中
getq():从任务的消息队列中取出消息
thr_count():返回任务中线程的数目
last_thread():返回任务中将线程计数器从1降为0的线程的ID
...
PS: 由于ACE_Task对象一般是在堆中创建的,因此必须要进行释放操作.
class CTaskDemo : public ACE_Task<ACE_MT_SYNCH>
{
public:
virtual int open (void *args = 0)
{
activate( THR_NEW_LWP, 1 );
return 0;
}
virtual int close (u_long flags = 0)
{
if ( ACE_OS::thr_equal ( ACE_Thread::self (), this->last_thread () ) )
{
//释放对象
delete this;
}
}
virtual int svc (void)
{
return 0;
}
};
{
public:
virtual int open (void *args = 0)
{
activate( THR_NEW_LWP, 1 );
return 0;
}
virtual int close (u_long flags = 0)
{
if ( ACE_OS::thr_equal ( ACE_Thread::self (), this->last_thread () ) )
{
//释放对象
delete this;
}
return 0;
}
virtual int svc (void)
{
return 0;
}
};