Task可以视为更高一个级别的线程对象，各个Task之间的同学也就是线程之间的通讯，在ACE中，采用了消息队列的方式进行处理。每一个Task封装一个ACE_Message_Queue，对于这个消息队列使用putq将消息压入队列，采用getq将消息出列，获取消息的顺序与压入消息的顺序相同。
下面的代码使用了消息队列在两个Task之间进行通讯，GenerateData任务生成数据，每间隔1秒钟会将全局变量g_nCounter增1，并将其压入队列。PrintTask任务从消息队列中获取消息，并将消息显示到输出设备上。
二者之间的通讯是通过PrintTask的消息队列实现。
#include "ace/Task.h" #include "ace/OS.h" int g_nCounter = 0; /*创建一个PrintTask类 *此类从消息队列中取出可用的消息，并将其值打印到输出设备 *在消息队列为空的情况下,getq进行阻塞处理 */ class PrintTask : public ACE_Task<ACE_MT_SYNCH> { public: //这个函数还在在原有的线程中调用的，即此时还未创建新的线程 int open(void*) { ACE_DEBUG((LM_DEBUG,ACE_TEXT("(%t)PrintTask:open/n"))); activate(THR_NEW_LWP); return 0; } //这是在新创建的子线程中调用的 int close(u_long flags) { ACE_DEBUG((LM_DEBUG,ACE_TEXT("(%t)PrintTask:close/n"))); return 0; } //一直从消息队列中获取消息 int svc(void) { ACE_DEBUG((LM_DEBUG,ACE_TEXT("(%t)PrintTask:svc/n"))); ACE_Message_Block* pMB; do { pMB = 0; getq(pMB); ACE_DEBUG((LM_DEBUG,ACE_TEXT("(%t) g_nCounter = %d/n"),*(int*)pMB->rd_ptr())); } while (1); return 0; } }; /* *生成数据任务 *此类的服务程序每间隔1秒钟将变量值递增1，并将其发送到消息队列 */ class GenerateData : public ACE_Task<ACE_MT_SYNCH> { public: GenerateData(PrintTask* pt) { m_pPrintTask = pt; m_pMB = new ACE_Message_Block((char*)&g_nCounter,sizeof(g_nCounter)); } //这个函数还在在原有的线程中调用的，即此时还未创建新的线程 int open(void*) { ACE_DEBUG((LM_DEBUG,ACE_TEXT("(%t)GenerateData:open/n"))); activate(THR_NEW_LWP); return 0; } //这是在新创建的子线程中调用的 int close(u_long flags) { ACE_DEBUG((LM_DEBUG,ACE_TEXT("(%t)GenerateData:close/n"))); return 0; } //想消息队列中放置一条消息，递增数值 int svc(void) { while(1) { m_pPrintTask->putq(m_pMB); g_nCounter++; ACE_OS::sleep(1); } return 0; } public: PrintTask* m_pPrintTask; ACE_Message_Block* m_pMB; }; int ACE_TMAIN(int argc, ACE_TCHAR* argv[]) { ACE_DEBUG((LM_DEBUG,ACE_TEXT("(%t)ACE_TMAIN/n"))); PrintTask pt; GenerateData gd(&pt); gd.open(0); pt.open(0); ACE_Thread_Manager::instance()->wait(); return 0; }
