ACE介绍.doc

ACE介绍一.摘要2二.适用者2三.开始23.1main23.2环境初始化23.3日志3四.内存使用42.1缓存内存分配类ACE_Cached_Allocator52.2内存分配器ACE_Malloc52.3内存映射：ACE_Mem_Map62.4内存池：ACE_XXX_Memory_Pool62.5共享内存：ACE_Shared_Memory_MM8五.锁93.1原子锁：ACE_Atomic_Op93.2其他锁：ACE_XXX_Mutex9六.队列93.3线性队列：ACE_Unbounded_Set93.4内存块模型：ACE_Data_Block103.5消息存储模型：ACE_Message_Block113.6消息队列：ACE_Message_Queue_Ex12七.线程125.1线程管理类：ACE_Thread_Manager12八.任务：ACE_Task123.1机制123.215九.网络通讯15十.反应器：ACE:Reactor15十一.ACE常用设计模式169.1单件模式之应用：Singleton169.2适配器模式之应用：Adapter179.3迭代器模式之应用：Iterator179.4置换模式之应用：Substitution189.5工厂模式18十二.主动对象介绍18十三.参考文献18一. 摘要ACE（自适配通信环境：Adaptive Communication Environment），封装了OS层的底层API，并结合设计模式为各种操作平台提供了一整套高性能的底层开发包。开发包的内容涉及到：进程间通信、内存管理、线程管理、同步、网络通信、事件分发、服务器配置等。二. 适合阅读者本文档适用ACE初学者，主要介绍如何利用ACE封装的优势来开发我们自己应用系统。 三. 开始3.1 mainACE对main做了重定义，目的是为了适应不同平台的编译入口点，这个一般不会影响到我们的开放过程，我们可以象平时那样按照main的格式写我们的入口函数：int main(int argv, char *argc);，main函数的两个参数一定要写，如果这样写:int main()会出现编译错误：unresolved external symbol int _cdecl ace_main_i(int,char * * const)。我们当然也可以不按照ACE规定的main格式，只要在main函数前面增加下面的语句就可以了：#ifdef main#undef main#endif3.2 环境初始化如果使用ACE类库提供的服务时例如：反应器（Reactor），需要在应用系统的开始和结束处各添加ACE:init()和ACE:fini()，分别用来初始化ACE环境和释放资源，另外对于单实例模式的使用也需要使用ACE:init和ACE:fini。这两个静态函数其实是调用ACE_Object_Manager类的init和fini方法。对于ACE提供的单实例模式类，为了统一封装并不要求用户自己释放，而是由ACE在通过ACE:fini统一在系统的结束处释放其资源。单实例的使用：3.3 日志ACE提供日志及跟踪信息的调试、打印管理，类ACE_Log_Msg提供日志的显示和打印，用户也可以通过继承ACE_Log_Msg_Callback类实现日志的自己管理。在我们代码里面可以使用两个宏：ACE_DEBUG和ACE_ERROR记录日志，下面例子简单显示两个日志：int main(int argc, char * argv)ACE_DEBUG(LM_DEBUG, “ACE_DEBUG测试n”); /注意宏前后一定要使用两个括号和日志相关的三个头文件：log_msg.h、log_record.h、log_msg_callback.h使用ACE_Log_Msg实现日志管理，默认情况下ACE_Log_Msg根据flags输出日志，默认的flags是ACE_Log_Msg:stderr，例如上面的例子日志被输出到console，我们可以通过修改flags改变日志输出的位置。1、 作为系统日志输出，下面的代码可以输出成系统日志，通过WINDOWS事件查看器可以看到结果。#include ace/log_msg.h#include ace/log_msg_callback.h#include ace/log_record.hint main (int argc, ACE_TCHAR * argv)/ACE_LOG_MSG宏其实是ACE_Log_Msg:instance()ACE_LOG_MSG-open(logtest, ACE_Log_Msg:SYSLOG); ACE_DEBUG(LM_INFO, 日志测试11n); 2、 输出到文件#include ace/Os.h#include ace/log_msg.h#include ace/log_msg_callback.h#include ace/log_record.h#include using namespace std;int main (int argc, ACE_TCHAR * argv)ACE_DEBUG(LM_DEBUG, first messagen);/清除原来的标志，如果不清除在输出到文件到同时也会在console上输出。ACE_LOG_MSG-clr_flags(ACE_Log_Msg:STDERR); ACE_DEBUG(LM_DEBUG, second messagen);ACE_LOG_MSG-set_flags(ACE_Log_Msg:OSTREAM);ACE_DEBUG(LM_DEBUG, third messagen);/定义文件名，并设置文件句柄const char * filename=main.log;ofstream out(filename, ios:out | ios:trunc);ACE_LOG_MSG-msg_ostream(&out);/测试信息，这次能真正输出到文件。ACE_DEBUG(LM_DEBUG, forth messagen);return 0;3、 自己处理日志#include ace/Os.h#include ace/log_msg.h#include ace/log_msg_callback.h#include ace/log_record.h/* * 自己处理日志类 */class CLogger : public ACE_Log_Msg_Callbackpublic:void log (ACE_Log_Record & record); /重定义虚函数;voidCLogger:log(ACE_Log_Record & record)/打印日志ACE_OS:printf(%sn, record.msg_data();int main (int argc, ACE_TCHAR * argv)ACE_DEBUG(LM_DEBUG, first messagen);ACE_LOG_MSG-clr_flags(ACE_Log_Msg:STDERR);ACE_DEBUG(LM_DEBUG, second messagen);ACE_LOG_MSG-set_flags(ACE_Log_Msg:MSG_CALLBACK);ACE_DEBUG(LM_DEBUG, third messagen);/设置自定义类CLogger logger;ACE_LOG_MSG-msg_callback(&logger);ACE_DEBUG(LM_DEBUG, forth messagen);return 0;四. 内存使用ACE的内存管理类常用的就是两个，一个是ACE_Cached_Allocator，一个是ACE_Malloc，这两个类都可以和锁相结合完成多线程中内存申请同步问题，下面介绍这两个类。2.1 缓存内存分配类ACE_Cached_Allocator这个类在构造函数中通过new操作预先分配一定的内存，并按照第一个模板参数的大小对内存进行分块把每一块的内存偏移指针作为链表的节点存放在成员变量ACE_Locked_Free_List中，在析构时释放已申请的内存。类的定义：ACE_Cached_Allocator ，第一个模板参数为链表节点内存块的大小，第二个模板参数为锁策略。用户常使用方法有malloc()、calloc()、free()，用户在调用这几个函数申请内存时，ACE_Cached_Allocator只是每次从链表中返回一个节点，用户每次申请内存的大小不能超过sizeof(T)，malloc和calloc不同之处是calloc根据第二个参数对内存初始化。下面为使用例子程序：#include ace/malloc.h struct SNodeint n;int main (int argc, ACE_TCHAR * argv)ACE_Cached_Allocator allocator(64);ACE_Cached_Allocator stu_allocator(8);char * p = (char*)allocator.malloc(32);if (!p) ACE_DEBUG(LM_DEBUG, 申请内存为空n);SNode * node = (SNode*)stu_allocator.malloc(sizeof(SNode);if (!node) ACE_DEBUG(LM_DEBUG, 申请内存为空n); else ACE_DEBUG(LM_DEBUG, val:%dn, node-n);stu_allocator.free(node);if ( node = (SNode*)stu_allocator.calloc(sizeof(SNode) ) ACE_DEBUG(LM_DEBUG, val:%dn, node-n);allocator.free(p);stu_allocator.free(node);return 0;2.2 内存分配器ACE_Malloc和ACE_Cached_Allocator不同的是，ACE_Malloc提供几个内存池类型，通过指定不同的内存池类型可以实现进程间内存共享，满足进程间内存共享的类型包括：ACE_MMAP_Memory_Pool、ACE_Shared_Memory_Pool、ACE_Local_Memory_Pool。ACE_Malloc只是根据第一个模板参数调用不同的内存池实现内存分配，在实现跨进程内存上ACE_Malloc通过一个名称与每次申请的内存块进行绑定，在进程内的使用函数有malloc、free，进程间多了bind、find函数，下面先演示进程内使用：#include ace/os.h#include ace/local_memory_pool.h#include ace/Synch.h#include ace/malloc.h#include ace/malloc_t.htypedef ACE_Malloc CMemMalloc;struct SNodeint n;int main(int argc, char * argv)CMemMalloc memMalloc;SNode * ptrNode = (SNode*)memMalloc.malloc ( sizeof(SNode) );if ( NULL = ptrNode ) ACE_DEBUG (LM_DEBUG, 申请内存失败n); else memMalloc.free (ptrNode);return 0;下面演示进程间使用：五. 线程5.1 简单的线程类ACE_ThreadACE_Thread实现了跨平台的线程封装，启动线程函数为静态函数ACE_Thread:spawn_n，通过join函数保证线程释放资源，按照OOP封装的原则并不建议用户过多的使用ACE_Thread，而应该采用后面介绍的ACE_Task，下面为简单的演示代码：#include ace/os.h#include ace/log_msg.h#include ace/thread.h#include ace/guard_t.h#include ace/thread_mutex.hstatic void Worker(void * arg)ACE_DEBUG(LM_DEBUG, 线程测试n);int main(int argc, char * argv)ACE_thread_t idThread;ACE_hthread_t hThread;if (-1 = ACE_Thread:spawn_n (&idThread,1,(ACE_THR_FUNC)Worker,0,THR_JOINABLE | THR_NEW_LWP,ACE_DEFAULT_THREAD_PRIORITY,0,0,&hThread) ACE_DEBUG(LM_DEBUG, starting threads failedn);return -1;getchar();ACE_Thread:join(hThread);return 0;5.2 高级线程管理类：ACE_Thread_ManagerACE_Thread_Manager是ACE_Thread的超集，增加了线程管理功能：启动、挂起、恢复、取消等；后面介绍的ACE_Task就是采用ACE_Thread_Manager管理自己的线程池。修改上面的代码如下：if ( -1 = ACE_Thread_Manager:instance()-spawn_n(1,(ACE_THR_FUNC)Worker,0,THR_NEW_LWP,ACE_DEFAULT_THREAD_PRIORITY,1) ACE_ERROR(LM_ERROR, Failure to spawn first group of threads: %p n);return -1;getchar();ACE_Thread_Manager:instance()-wait_grp(1);六. 锁3.1 线程锁ACE_Thread_Mutex线程锁在WIN32下为临界区CRITICAL_SECTION，其他平台下调用pthread_mutex_lock实现了跨平台的通用锁，通过锁的函数为acquire并可以传递参数设置超时，当acquire超时时返回-1并且errno等于ETIME，tryacquire函数是一个测试是否可以获取锁的函数-1失败并设置errno为EBUSY，释放锁函数release，下面为使用方法：#include ace/os.h#include ace/log_msg.h#include ace/thread.h#include ace/thread_mutex.h#define THREAD_NUMS 2class CMutexTestpublic:CMutexTest():_bRun(0)int Open ();void Close ();private:friend void Worker(void *arg);private:bool _bRun;ACE_Thread_Mutex _Mutex;ACE_thread_t _ThreadIdsTHREAD_NUMS;ACE_hthread_t _ThreadHandleTHREAD_NUMS;intCMutexTest:Open ()if (-1 = ACE_Thread:spawn_n (_ThreadIds,THREAD_NUMS,(ACE_THR_FUNC)Worker,this,THR_JOINABLE | THR_NEW_LWP,ACE_DEFAULT_THREAD_PRIORITY,0,0,_ThreadHandle) ACE_DEBUG(LM_DEBUG, starting threads failedn);return -1;_bRun = 1;return 0;void Worker(void * arg)CMutexTest * pMutex = (CMutexTest*)arg;ACE_Time_Value t(0, 10);for (int i = 0; i _ThreadIdsi ) while (pMutex-_bRun ) pMutex-_Mutex.acquire ();ACE_DEBUG(LM_DEBUG, enter thread:%dn, pMutex-_ThreadIdsi);pMutex-_Mutex.release ();ACE_OS:sleep(t);ACE_DEBUG(LM_DEBUG, exit thread:%dn, pMutex-_ThreadIdsi);void CMutexTest:Close()_bRun = 0;for (int i = 0; i THREAD_NUMS; i+)ACE_Thread:join (_ThreadHandlei);int main (int argc, char * argv)CMutexTest mutex;if (mutex.Open () != -1) getchar ();mutex.Close ();return 0;3.2 读写线程锁 ACE_RW_Thread_Mutex读写锁的机制是可允许多个线程同时对一个数据进行“读”操作，一个“写”操作和多个“读”操作之间是互斥的。WIN32下没有提供专门的读写锁，在*NIX下有对应的读写锁；操作函数有：acquire_write、acquire_read、release。3.3 原子锁：ACE_Atomic_Op定义为templateclass ACE_Export ACE_Atomic_Op，相当于WIN32的InterLockedIncrement()和InterLockedDecrement()函数，由于此类内部实现了operator+、operator-、等操作，使用起来比较方便。例如：ACE_Atomic_Op val;val+; /递增val-; /递减int I = val.value(); /返回当前值val = 5; /可以直接赋值3.4 锁守护：ACE_Guard为了防止程序员忘记释放锁，往往通过类的构造和析构函数特性实现自动加锁和释放锁，在ACE里面也提供这样的机制ACE_Guard，使用方法如下：void Worker(void * arg)CMutexTest * pMutex = (CMutexTest*)arg;ACE_Time_Value t(0, 10);for (int i = 0; i _ThreadIdsi ) while (pMutex-_bRun ) ACE_Guard guard (pMutex-_Mutex);ACE_DEBUG(LM_DEBUG, enter thread:%dn, pMutex-_ThreadIdsi);ACE_OS:sleep(t);ACE_DEBUG(LM_DEBUG, exit thread:%dn, pMutex-_ThreadIdsi);七. 队列3.5 线性队列：ACE_Unbounded_SetACE_Unbounded_Set使用ACE_NODE作为自己的链表节点，提供的访问方法有：insert、find、remove、begin、end、copy_nodes、delete_nodes。ACE_Unbounded_Set通过自己的迭代器对成员进行遍历操作。Insert：向队列添加一个数据成员，队列通过malloc构造成员。Find：在队列查找指定的成员，0成功，-1失败。Remove：在队列移出指定的成员。0成功，-1失败Begin：返回队列的开始节点。End：返回队列的尾节点Copy_node：把一个队列的所有节点添加到本队列的末尾。Delete_node：删除本队列的所有节点。ACE_Unbounded_Set没有提供锁保护用户在外部使用时需要采用合适的锁。例子：int main(int argv, char * argc)ACE_Unbounded_Set liner;ACE_Unbounded_Set_Iterator it(liner);for (int i = 0; i 10; i+)liner.insert(i*100);printf(size of liner:%dn, liner.size();for (it = liner.begin (); it != liner.end(); it+)printf(%d , *it);liner.remove(500);printf(nafter remove 500:n);for (it = liner.begin (); it != liner.end(); it+)printf(%d , *it);if (liner.find(300) = 0)printf(nthe element 300 has been found:n);return 0;3.6 内存块模型：ACE_Data_BlockACE_Data_Block功能简单描述就是通过锁策略和共享引用计数管理一块内存。提供两个内存分配接口allocator_strategy_和data_block_allocator_， 外部的使用者例如ACE_Message_Block可以利用ACE_Data_Block实现数据管理。ACE_Data_Block的构造函数定义如下：ACE_Data_Block:ACE_Data_Block (void); /缺省的构造函数ACE_Data_Block:ACE_Data_Block (size_t size, ACE_Message_Block:ACE_Message_Type msg_type, const char *msg_data, ACE_Allocator *allocator_strategy, ACE_Lock *locking_strategy, ACE_Message_Block:Message_Flags flags, ACE_Allocator *data_block_allocator);下面是常用函数的演示程序：int main(int argv, char * argc)ACE_Lock_Adapter mutex;ACE_Data_Block block(32, ACE_Message_Block:MB_DATA, 0, 0, &mutex, 0, 0);ACE_Data_Block * p_b, *p_b2;int i;printf(size of block:%dn, block.size();printf(capacity of block:%dn, block.capacity();int * p = (int*)block.base();for (i = 0; i base();for (i = 0; i size()/4; i+)printf(%d , *p+);if (p_b = &block & p_b-base() = block.base()printf(the two blocks is equaln);p_b2 = block.clone(0);p = (int *)p_b2-base();for (i = 0; i size()/4; i+)printf(%d , *p+);if (p_b2 != p_b & p_b2-base() != p_b-base()printf(the two blocks is not equaln);p_b2-release();return 0;3.7 消息存储模型：ACE_Message_BlockACE_Message_Block的作用是在ACE_Data_Block基础上进行数据管理，可以在不同的内存块标注不同的数据类型，提供数据的读取和写入操作，通过prev和next指针实现Double Link。ACE_Message_Block主要为ACE_Message_Block队列提供底层机制。/简单初始化创建空消息ACE_Message_Block (ACE_Allocator *message_block_allocator = 0);ACE_Message_Block:ACE_Message_Block (const char *data, size_t size, unsigned long priority)3.8 消息队列：ACE_Message_Queue_ExACE_Message_Queue_Ex是一个功能更强的消息队列，把ACE_Message_Queue作为成员变量实现更丰富的功能，使用者在外面调用enqueue_tail、dequeue_head等函数时不再象ACE_Message_Queue那样需要传递ACE_Message_Block指针，从而实现了更好的封装。l 几行最简单的代码：#include ace/os.h#include ace/condition_thread_mutex.h#include ace/thread.h#include ace/message_queue_t.hstruct SNodeSNode(int m) : _n(m)int _n;typedef ACE_Message_Queue_Ex MessageQueue;int main(int argc, char * argv)MessageQueue msgQueue;SNode * pNode = NULL;for (int i = 0; i _n);delete pNode;return 0;l 水位的作用在ACE_Message_Queue_Ex队列的构造函数或者open函数中需要传递两个水位标志（低水位标志和高水位标志），用于在流中的相邻Module间实现层到层的流控制。高水位标指示消息队列在进行流控制之前所愿意缓存的消息字节数。低水位标指示在先前已进行流控制的Message Queue不再被视为满的“水位”。l 队列的通知八. 任务：ACE_Task3.1 原理ACE_Task通过把多线程和队列封装实现了一种称为主动对象的并发编程模式。传统的“被动对象”是请求到达个对象时它的处理才会被激活，每个对象和其后继者之间关系比较紧耦合（因为请求的方法调用和请求者位于相同的线程内）。主动对象模式是通过多线程循环主动去检测队列上的数据，并能对队列中的请求进行并发处理而不会阻塞；ACE_Task提供了实现这种模式的机制，ACE_Task启动函数：open，在open虚函数内可以通过activate函数激活相应数量的线程池，用户代码可以在外部向ACE_Task队列插入请求。ACE_Task类中的队列采用ACE_Message_Queue队列，另外一个扩展版本类ACE_Task_Ex采用了ACE_Message_Queue_Ex队列。Open虚函数在ACE_Task类内部并没有做任何事情，作为ACE的习惯open函数基本上时所有类的起始函数，用户程序可以在此函数内部完成初始化工作，比如调用activate启动线程。Activate函数定义如下：activate (long flags = THR_NEW_LWP | THR_JOINABLE |THR_INHERIT_SCHED , int n_threads = 1, int force_active = 0, long priority = ACE_DEFAULT_THREAD_PRIORITY, int grp_id = -1, ACE_Task_Base *task = 0, ACE_hthread_t thread_handles = 0, void *stack = 0, size_t stack_size = 0, ACE_thread_t thread_ids = 0);这个函数参数看上去比较多，其实大部分都可以使用缺省参数，用户在调用时可以只调用activate()或者activate(LWP_NEW_THR, 1)。如果需要启动多个线程时我们可以这样调用activate(LWP_NEW_THR, 10, 0, ACE_DEFAULT_THREAD_PRIORITY, -1, 0, 0, 0, 0, thread_ids)，最后一个参数thread_ids在多线程中时有用的。ACE_Task线程运行函数：svcsvc函数是真正的线程函数体，svc也是虚函数需要用户重置，当通过activate启动多个线程后，我们可以在svc函数体内根据ACE_Thread_:self()=thread_idsi 比较来启动相应的用户处理函数。例如：#define THREADS 2class Task_Test : public ACE_Taskpublic:Task_Test();int open(void*);int svc(void);int close(u_long);private:void handle_thread_1(u_long);void handle_thread_2(u_long);private:ACE_thread_t thread_idsTHREADS;Task_Test:Task_Test()wait(); /等待所有的线程退出int Task_Test:open(void*)activate(THR_NEW_LWP,THREADS,0,ACE_DEFAULT_THREAD_PRIORITY,-1,0,0,0,0,thread_ids);return 0;int Task_Test:svc(void)for (int i = 0; i THREADS; i+) if (ACE_Thread:self() = thread_idsi)switch (i) case 0:handle_thread_1(thread_idsi);break;case 1:handle_thread_2(thread_idsi);break;default:break;return 0;int Task_Test:close(u_long code)printf(thread exit:%dn, code);return 0;void Task_Test:handle_thread_1(u_long code)printf(this is thread:%dn, code);void Task_Test:handle_thread_2(u_long code)printf(this is thread:%dn, code);int main(int argv, char * argc)Task_Test task;task.open(0);return 0;ACE_Task内置队列：ACE_Message_Queue，下面演示如何使用队列：#include ace/os.h#include ace/thread_mutex.h#include ace/task.h#include ace/malloc_t.hstruct SNodeint _n;class CServer;class CClient /发送请求者public:CClient (CServer * pSvr) : _pServer(pSvr) void SendMsg (SNode *);private:CServer * _pServer;class CServer : public ACE_Task /线程安全public:CServer()wait();int open(void *);int svc(void);private:int close (u_long lFlag);/* *客户端处理函数 */void CClient:SendMsg(SNode * pNode)ACE_Message_Block * mb = new ACE_Message_Block(char*)pNode, sizeof(SNode);if (_pServer)_pServer-putq(mb);/* * 服务端代码 */int CServer:open (void *)activate(THR_NEW_LWP, 1);return 0;int CServer:svc()ACE_Message_Block * mb = NULL;while ( 1 ) getq(mb);if (mb-rd_ptr() = (char*)-1)break;ACE_DEBUG(LM_DEBUG, get value:%dn, *mb-rd_ptr();delete mb;return 0;int CServer:close(u_long lFlag)ACE_DEBUG(LM_DEBUG, 线程退出n);return 0;int main(int argc, char * argv)ACE_Malloc memMalloc;CServer svr;CClient cli(&svr);if (svr.open (0) = -1)ACE_DEBUG(LM_DEBUG, 启动线程失败n);elsefor (int i =