ACE_Proactor网络通信示例

2023-12-06 20:18
文章标签 示例 网络通信 ace proactor

本文主要是介绍ACE_Proactor网络通信示例，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

注：本文仅对使用ACE进行网络通信进行演示说明。本文中的代码皆使用doxgen的注释风格。本文中使用的事件机制，其原理与实现请参考[ 基于C++的事件机制设计［2.0］]一文。

 

ACE的Proactor对Epoll和IOCP进行了良好包装，因此，使用ACE来进行网络开发是相当的便利，性能也不差。闲言少叙，看代码。

这里以TCP协议进行流式通信。我们需要解析流，得出每次接收到的数据包大小和包含的数据域，假定我们的包结构如下：

 

包序列号(32Bit) | 长度(16Bit) | 数据域(大小为长度所表示的字节)... | (下一包)

 

通过分析由包序列号和长度组成的包头来解决半包，粘包等问题，许多其它文章也有描述，这里就省略了。

这样可以确定我们的包头结构如下：

 

[cpp] view plain copy
  1. #pragma pack(push)  
  2. #pragma pack(1)  
  3.     /** 
  4.     * @brief Tcp包头结构 
  5.     */  
  6.     typedef struct tag_TTcpPackHeader  
  7.     {  
  8.         unsigned int seq; //< 包序号  
  9.         unsigned short len; //< 包长度  
  10.     }TTcpPackHeader;  
  11. #pragma pack(pop)  
  12. /// 包头尺寸宏  
  13. #define TCP_PACK_HEADER_SIZE sizeof(tag_TTcpPackHeader)  
.

需要注意的是，要求在字节边界对齐。

 

现在来看看通过ACE来实现TCP通信需要哪些东西：

INET_Addr 用于地址访问

Task_Base 用于线程模型

Message_Block 用于消息传递和数据容器

Asynch_IO 异步通信

Proactor IOCP架构

 

并且，要建立这样的通信架构，我们需要：

一个Acceptor：用于接受连接

一个Handler：对应于每个连接句柄，并用于数据的发送/接收。

一个事件分发线程：以事件的形式将接收到数据分发出去，并在对应的句柄上进行数据发送。

 

本示例并没有采用在接收到数据时立即进行处理的方式，而是通过创建一个额外的事件分发线程的形式，将数据包投递到该线程的消息队列中，由该线程向外派送。因此，数据处理与网络层是隔离的，且网络层能专注于通信，最大的发挥效用。

 

好了，下面来看看实现：

先看Handler，参考ACE_Service_Handler，我们需要重载open()，addresses(), handle_read_stream()，handle_write_stream()，以在连接打开时进行读写流对象的初始化、获取客户端地址，处理输入/输入流。

 

注：以T作为类的开头而不是C，是出于对曾经伟大的BORLAND的深刻怀念。

注：成员又以m_开头，是出于对现而今仍伟大的MS的深刻怨念。

 

[cpp] view plain copy
  1. /** 
  2. * @class TTcpHandler 
  3. * @brief Tcp连接句柄 
  4. */  
  5. class TTcpHandler : public ACE_Service_Handler  
  6. {  
  7. public:  
  8.     /** 
  9.     * @brief 客户端连接事件类型定义 
  10.     * @param [in] ACE_UINT32 客户端地址 
  11.     * @param [in] ACE_UINT16 客户端端口 
  12.     * @param [in] TTcpHandler* 连接句柄 
  13.     */  
  14.     typedef TEvent<void, ACE_UINT32, ACE_UINT16, TTcpHandler *> TOnClientConnect;  
  15.     /** 
  16.     * @brief 客户端断开连接事件类型定义 
  17.     * @param [in] ACE_UINT32 客户端地址 
  18.     * @param [in] ACE_UINT16 客户端端口 
  19.     */  
  20.     typedef TEvent<void, ACE_UINT32, ACE_UINT16> TOnClientDisconnect;  
  21.     /** 
  22.     * @brief 客户端连接验证事件 
  23.     * @param [in] ACE_UINT32 客户端地址 
  24.     * @param [in] ACE_UINT16 客户端端口 
  25.     * @return bool 
  26.     * - true 验证通过 
  27.     * - false 验证失败 
  28.     */  
  29.     typedef TEvent<bool, ACE_UINT32, ACE_UINT16> TOnClientValidate;  
  30.     /** 
  31.     * @brief 接收到客户端数据事件类型定义 
  32.     * @param [in] ACE_UINT32 客户端地址 
  33.     * @param [in] ACE_UINT16 客户端端口 
  34.     * @param [in] unsigned int 数据包序列号 
  35.     * @param [in] const char* 数据区域指针 
  36.     * @param [in] size_t 数据长度 
  37.     */  
  38.     typedef TEvent<void, ACE_UINT32, ACE_UINT16, unsigned int, const char*, unsigned short> TOnDataReceive;  
  39.       
  40.     /** 
  41.     * @brief 成功发送客户端数据事件类型定义 
  42.     * @param [in] ACE_UINT32 客户端地址 
  43.     * @param [in] ACE_UINT16 客户端端口 
  44.     * @param [in] unsigned int 数据包序列号 
  45.     * @param [in] const char* 数据区域指针 
  46.     * @param [in] size_t 数据长度 
  47.     */  
  48.     typedef TEvent<void, ACE_UINT32, ACE_UINT16, unsigned int, const char*, unsigned short> TOnDataSendSucceeded;  
  49.       
  50.     /** 
  51.     * @brief 失败发送客户端数据事件类型定义 
  52.     * @param [in] ACE_UINT32 客户端地址 
  53.     * @param [in] ACE_UINT16 客户端端口 
  54.     * @param [in] unsigned int 数据包序列号 
  55.     * @param [in] const char* 数据区域指针 
  56.     * @param [in] size_t 数据长度 
  57.     */  
  58.     typedef TEvent<void, ACE_UINT32, ACE_UINT16, unsigned int, const char*, unsigned short> TOnDataSendFailed;  
  59. private:  
  60.     ACE_Asynch_Read_Stream m_Reader; //< 异步读数据流  
  61.     ACE_Asynch_Write_Stream m_Writer; //< 异步写数据流  
  62.     ACE_Message_Block* m_CurDataMB; //< 当前读取数据  
  63.     ACE_INET_Addr m_ClientAddr; //< 客户端地址  
  64. public:  
  65.     /** 
  66.     * @name 事件句柄 
  67.     * @{ 
  68.     */  
  69.     DECL_PROP(TOnClientConnect, OnClientConnect) //< 客户端连接事件句柄  
  70.     DECL_PROP(TOnClientDisconnect, OnClientDisconnect) //< 客户端断开事件句柄  
  71.     DECL_PROP(TOnDataReceive, OnDataReceive) //< 接收到数据的事件句柄  
  72.     DECL_PROP(TOnDataSendSucceeded, OnDataSendSucceeded) //< 成功发送数据的事件句柄  
  73.     DECL_PROP(TOnDataSendFailed, OnDataSendFailed) //< 发送数据失败的事件句柄  
  74.     /** 
  75.     * @} 
  76.     */  
  77. public:  
  78.     /// ctor  
  79.     TTcpHandler();  
  80.       
  81.     /// dtor  
  82.     ~TTcpHandler();  
  83.       
  84.     /** 
  85.     * @brief 发送数据 
  86.     * @param [in] unsigned int 数据包序列号 
  87.     * @param [in] const char* 要发送的数据区域指针 
  88.     * @param [in] size_t 要发送的数据长度 
  89.     * @return int 
  90.     * - 0 成功 
  91.     * - 1 失败 
  92.     */  
  93.     int send(unsigned int seq, const char* data, unsigned short size);  
  94.       
  95.     /** 
  96.     * @brief 打开句柄 
  97.     * @see ACE_Service_Handler 
  98.     */  
  99.     virtual void open(ACE_HANDLE h, ACE_Message_Block& mb);  
  100.     /** 
  101.     * @brief 获取地址 
  102.     * @see ACE_Service_Handler 
  103.     */  
  104.     virtual void addresses (const ACE_INET_Addr &remote_address,  
  105.                           const ACE_INET_Addr &local_address);  
  106.       
  107.     /** 
  108.     * @brief 读取流数据 
  109.     * @see ACE_Service_Handler 
  110.     */  
  111.     virtual void handle_read_stream(const ACE_Asynch_Read_Stream::Result& result);  
  112.     /** 
  113.     * @brief 写入流数据 
  114.     * @see ACE_Service_Handler 
  115.     */  
  116.     virtual void handle_write_stream(const ACE_Asynch_Write_Stream::Result& result);  
  117.     /** 
  118.     * @brief 初始化当前数据接收缓冲事件 
  119.     */  
  120.     void initCurDataMB();  
  121. }; // class TTcpHandler  

 

而相应滴，Acceptor在接受连接时，产生出的Handler应该是TTcpHandler类型，其定义如下：

注意，为了将事件句柄与连接句柄（TTcpHandler）挂钩，这里重载了make_handler()。而重载validate_connection则是为了让连接验证事件能够在恰当的时机被激发。

[cpp] view plain copy
  1. /** 
  2. * @class TTcpAcceptor 
  3. * @brief TCP接受器 
  4. * @see ACE_Asynch_Acceptor 
  5. * @see TTcpHandler 
  6. */  
  7. class TTcpAcceptor : public ACE_Asynch_Acceptor<TTcpHandler>  
  8. {  
  9. public:  
  10.     /** 
  11.     * @name TCP事件句柄 
  12.     * @see TTcpHandler 
  13.     * @{ 
  14.     */  
  15.     DECL_PROP(TTcpHandler::TOnClientConnect, OnClientConnect)  
  16.     DECL_PROP(TTcpHandler::TOnClientDisconnect, OnClientDisconnect)  
  17.     DECL_PROP(TTcpHandler::TOnClientValidate, OnClientValidate)  
  18.     DECL_PROP(TTcpHandler::TOnDataReceive, OnDataReceive)  
  19.     DECL_PROP(TTcpHandler::TOnDataSendSucceeded, OnDataSendSucceeded)  
  20.     DECL_PROP(TTcpHandler::TOnDataSendFailed, OnDataSendFailed)  
  21.     /** 
  22.     * @} 
  23.     */  
  24. protected:  
  25.     /** 
  26.     * @brief 连接验证 
  27.     * @note 激发 OnClientValidate 事件 @see TOnClientValidate 
  28.     * @see ACE_Asynch_Acceptor 
  29.     */  
  30.     virtual int validate_connection (const ACE_Asynch_Accept::Result& result,  
  31.                                const ACE_INET_Addr &remote,  
  32.                                const ACE_INET_Addr& local);  
  33.     /** 
  34.     * @brief 创建连接句柄事件 
  35.     * @see ACE_Asynch_Acceptor 
  36.     */  
  37.     virtual TTcpHandler* make_handler(void);  
  38. }; // class TTcpAcceptor  

 

有了Acceptor和Handler，还需要使之运行于Proactor模式下，因此有了以下线程：

[cpp] view plain copy
  1. /** 
  2. * @class TTcpNetThread 
  3. * @brief TCP网络线程 
  4. * @see ACE_Task_Base 
  5. * @see ACE_Proactor 
  6. */  
  7. class TTcpNetThread : public ACE_Task_Base  
  8. {  
  9. public:  
  10.     /** 
  11.     * @name TCP事件句柄 
  12.     * @see TTcpHandler 
  13.     * @{ 
  14.     */  
  15.     DECL_PROP(TTcpHandler::TOnClientConnect, OnClientConnect)  
  16.     DECL_PROP(TTcpHandler::TOnClientDisconnect, OnClientDisconnect)  
  17.     DECL_PROP(TTcpHandler::TOnClientValidate, OnClientValidate)  
  18.     DECL_PROP(TTcpHandler::TOnDataReceive, OnDataReceive)  
  19.     DECL_PROP(TTcpHandler::TOnDataSendSucceeded, OnDataSendSucceeded)  
  20.     DECL_PROP(TTcpHandler::TOnDataSendFailed, OnDataSendFailed)  
  21.     /** 
  22.     * @} 
  23.     */  
  24.     /// 运行  
  25.     int open();  
  26.     /// 停止运行  
  27.     int close();  
  28. protected:  
  29.     /// 线程函数  
  30.     virtual int svc();  
  31. };  

 

最后再看看事件分发线程，该线程也是对上述实现的聚合和封装，对外暴露事件和发送方法：

注意，该类也负责响应TTcpNetThread所激发的事件，所以需要派生自TObject。

 

[cpp] view plain copy
  1. /** 
  2. * @class TTcp 
  3. * @brief TCP接收和事件处理代理线程 
  4. */  
  5. class TTcp : public TObject, public ACE_Task<ACE_MT_SYNCH>  
  6. {  
  7. public:  
  8.     /** 
  9.     * @name 重定义事件类型 
  10.     * @see TTcpHandler 
  11.     * @{ 
  12.     */  
  13.     typedef TTcpHandler::TOnClientConnect TOnClientConnect;  
  14.     typedef TTcpHandler::TOnClientDisconnect TOnClientDisconnect;  
  15.     typedef TTcpHandler::TOnClientValidate TOnClientValidate;  
  16.     typedef TTcpHandler::TOnDataReceive TOnDataReceive;  
  17.     typedef TTcpHandler::TOnDataSendSucceeded TOnDataSendSucceeded;  
  18.     typedef TTcpHandler::TOnDataSendFailed TOnDataSendFailed;  
  19.     /** 
  20.     * @} 
  21.     */  
  22. private:  
  23.     /** 
  24.     * @name 成员变量 
  25.     * @{ 
  26.     */  
  27.     ACE_Recursive_Thread_Mutex m_Lock; //< 线程锁  
  28.     hash_map<unsigned __int64, TTcpHandler *> m_AddrMap; //< 地址/句柄映射  
  29.     TTcpNetThread* m_TcpNetThd;  
  30.     /** 
  31.     * @} 
  32.     */  
  33. public:  
  34.     /** 
  35.     * @name TCP事件句柄 
  36.     * @see TTcpHandler 
  37.     * @{ 
  38.     */  
  39.     DECL_PROP(TTcpHandler::TOnClientConnect, OnClientConnect)  
  40.     DECL_PROP(TTcpHandler::TOnClientDisconnect, OnClientDisconnect)  
  41.     DECL_PROP(TTcpHandler::TOnClientValidate, OnClientValidate)  
  42.     DECL_PROP(TTcpHandler::TOnDataReceive, OnDataReceive)  
  43.     DECL_PROP(TTcpHandler::TOnDataSendSucceeded, OnDataSendSucceeded)  
  44.     DECL_PROP(TTcpHandler::TOnDataSendFailed, OnDataSendFailed)  
  45.     /** 
  46.     * @} 
  47.     */  
  48. public:  
  49.     /// ctor  
  50.     TTcp();  
  51.     /// dtor  
  52.     ~TTcp();  
  53.     /// 运行  
  54.     void open();  
  55.     /// 停止  
  56.     void close();  
  57.       
  58.     /// 发送数据  
  59.     int send(ACE_UINT32 ip, ACE_UINT16 port, unsigned int seq, const char* buf, unsigned short len);  
  60. private:  
  61.     /// 线程函数  
  62.     virtual int svc();  
  63. private:  
  64.     /** 
  65.     * @name TTcpNetThread 事件处理方法 
  66.     * 关于事件原型的定义，请参考 @see TTcpHandler 
  67.     * @{ 
  68.     */  
  69.     void tcpNetThread_OnClientConnect(ACE_UINT32 ip, ACE_UINT16 port, TTcpHandler* handler);  
  70.     void tcpNetThread_OnClientDisconnect(ACE_UINT32 ip, ACE_UINT16 port);  
  71.     void tcpNetThread_OnDataReceive(ACE_UINT32 ip, ACE_UINT16 port, unsigned int seq, const char* data, unsigned short size);  
  72.     void tcpNetThread_OnDataSendSucceeded(ACE_UINT32 ip, ACE_UINT16 port, unsigned int seq, const char* data, unsigned short size);  
  73.     void tcpNetThread_OnDataSendFailed(ACE_UINT32 ip, ACE_UINT16 port, unsigned int seq, const char* data, unsigned short size);  
  74.     /** 
  75.     * @} 
  76.     */  
  77. }; // class TTcp  

 

现在基本格调已经确定，需要做的是编写具体实现代码了。

 

 

此乃末技。

 

 

应用ACE来作为底层通信的框架，已经是许多年前的技术了，这里纯粹是凑字数，骗更新滴。这样的老东西，确实是相当的让人无语。

 现在我们一步步来看看实现：

先是TTcpAcceptor，该类仅重载了两个方法，如下：

[cpp] view plain copy
  1. #include "TCPAcceptor.h"  
  2. namespace igame  
  3. {  
  4.     int TTcpAcceptor::validate_connection (const ACE_Asynch_Accept::Result& result,  
  5.                                        const ACE_INET_Addr &remote,  
  6.                                        const ACE_INET_Addr& local)  
  7.     {  
  8.         if (m_OnClientValidate.valid())  
  9.             // 这里激发TOnClientValidate事件  
  10.             return m_OnClientValidate(remote.get_ip_address(), remote.get_port_number()) ? 0 : -1;  
  11.         else  
  12.             return 0; // 默认允许连接  
  13.     }  
  14.     TTcpHandler* TTcpAcceptor::make_handler(void)  
  15.     {  
  16.         TTcpHandler* handler = 0;  
  17.         ACE_NEW_RETURN (handler, TTcpHandler(), 0);  
  18.         // 设置事件句柄  
  19.         handler->setOnClientConnect(m_OnClientConnect);  
  20.         handler->setOnClientDisconnect(m_OnClientDisconnect);  
  21.         handler->setOnDataReceive(m_OnDataReceive);  
  22.         handler->setOnDataSendSucceeded(m_OnDataSendSucceeded);  
  23.         handler->setOnDataSendFailed(m_OnDataSendFailed);  
  24.         return handler;  
  25.     }  
  26. } // namespace igame  

 

复杂的部分在TTcpHandler，该类不仅需要接收数据（拼包），也要处理发送：

 

[cpp] view plain copy
  1. #include "TcpHandler.h"  
  2. namespace igame  
  3. {  
  4.     TTcpHandler::TTcpHandler()  
  5.         :m_CurDataMB(0) // 初始化  
  6.     { }  
  7.     TTcpHandler::~TTcpHandler()  
  8.     {  
  9.         if (handle() != ACE_INVALID_HANDLE)  
  10.         {  
  11.             ACE_OS::closesocket(handle()); // 关闭句柄  
  12. #ifdef _DEBUG  
  13.             // 打印调试信息  
  14.             ACE_TCHAR remoteAddrStr[128];  
  15.               
  16.             m_ClientAddr.addr_to_string(remoteAddrStr, sizeof(remoteAddrStr) / sizeof(ACE_TCHAR));  
  17.             ACE_DEBUG((LM_INFO, ACE_TEXT("Disconnect from %s/n"), remoteAddrStr));  
  18. #endif  
  19.             // 客户端断开  
  20.             m_OnClientDisconnect(m_ClientAddr.get_ip_address(), m_ClientAddr.get_port_number());  
  21.             if (m_CurDataMB)  
  22.                 m_CurDataMB->release();  
  23.         }  
  24.     }  
  25.     int TTcpHandler::send(unsigned int seq, const char* data, unsigned short dataSize)  
  26.     {  
  27.         ACE_Message_Block* dataMB = 0;  
  28.           
  29.         ACE_NEW_NORETURN(dataMB, ACE_Message_Block(sizeof(unsigned int) + sizeof(unsigned short) + dataSize));  
  30.           
  31.         short len = dataSize;  
  32.         dataMB->copy((const char *)&seq, sizeof(unsigned int)); // 这里没有处理seq  
  33.         dataMB->copy((const char *)&len, sizeof(unsigned short));  
  34.         dataMB->copy((const char *)data, dataSize);  
  35.               
  36.         int ret = m_Writer.write(*dataMB, dataMB->length()); // 发送  
  37.         if (ret == -1)  
  38.             m_OnDataSendFailed(m_ClientAddr.get_ip_address(), m_ClientAddr.get_port_number(), seq, data, dataSize);  
  39.         else  
  40.             m_OnDataSendSucceeded(m_ClientAddr.get_ip_address(), m_ClientAddr.get_port_number(), seq, data, dataSize);  
  41.         return ret;  
  42.     }  
  43.     void TTcpHandler::addresses (const ACE_INET_Addr &remote_address,  
  44.                           const ACE_INET_Addr &local_address)  
  45.     {  
  46.         m_ClientAddr = remote_address; // 取得客户端地址  
  47.     }  
  48.     void TTcpHandler::open(ACE_HANDLE h, ACE_Message_Block& mb)  
  49.     {  
  50.         handle(h); // set handle  
  51.           
  52.         if (m_Reader.open(*this) == -1) // 允许读  
  53.         {  
  54.             ACE_ERROR((LM_ERROR, ACE_TEXT("failed to open read handle %i/n"), errno));  
  55.             delete this;  
  56.             return;  
  57.         }  
  58.         if (m_Writer.open(*this) == -1) // 允许写  
  59.         {  
  60.             ACE_ERROR((LM_ERROR, ACE_TEXT("failed to open write handle %i/n"), errno));  
  61.             delete this;  
  62.             return;  
  63.         }  
  64.           
  65.         // 激发客户端连接事件  
  66.         m_OnClientConnect(m_ClientAddr.get_ip_address(), m_ClientAddr.get_port_number(), this);  
  67.         initCurDataMB();  
  68.           
  69.         m_Reader.read(*m_CurDataMB, m_CurDataMB->space()); // 读数据  
  70.     }  
  71.     void TTcpHandler::handle_read_stream(const ACE_Asynch_Read_Stream::Result& result)  
  72.     {  
  73.         ACE_Message_Block& mb = result.message_block();  
  74.         if (!result.success() || result.bytes_transferred() == 0) // no data or failed?  
  75.         {  
  76.             mb.release();  
  77.             delete this;  
  78.         }  
  79.         else  
  80.         {  
  81.             if (this->m_CurDataMB->length() < TCP_PACK_HEADER_SIZE) // try to read header info  
  82.             {  
  83.                 this->m_Reader.read(*m_CurDataMB, m_CurDataMB->space());  
  84.                 return ;  
  85.             }  
  86.             TTcpPackHeader* header = reinterpret_cast<TTcpPackHeader *>(this->m_CurDataMB->rd_ptr());  
  87.             ACE_Message_Block* dataMB = this->m_CurDataMB->cont();  
  88.             if (!dataMB)  
  89.             {  
  90.                 ACE_NEW_NORETURN(dataMB, ACE_Message_Block(header->len));  
  91.                 if (dataMB)  
  92.                     this->m_CurDataMB->cont(dataMB);  
  93.                 else  
  94.                 {  
  95.                     this->m_CurDataMB->release();  
  96.                     ACE_DEBUG((LM_ERROR, ACE_TEXT("Failed to allocated: %i/n"), errno));  
  97.                     delete this;  
  98.                     return ;  
  99.                 }  
  100.             }  
  101.               
  102.             if (dataMB->length() == header->len)  
  103.             {  
  104.                 // 成功读取了数据？  
  105.                 m_OnDataReceive(m_ClientAddr.get_ip_address(), m_ClientAddr.get_port_number(), header->seq, dataMB->rd_ptr(), header->len);  
  106.                   
  107.                 m_CurDataMB->release();  
  108.                 initCurDataMB(); // 下一包数据  
  109.                   
  110.                 this->m_Reader.read(*m_CurDataMB, m_CurDataMB->space()); // next, try to get header  
  111.                 return ;  
  112.             }  
  113.               
  114.             this->m_Reader.read(*dataMB, dataMB->space()); // try to get data left  
  115.         }  
  116.     }  
  117.     void TTcpHandler::handle_write_stream(const ACE_Asynch_Write_Stream::Result& result)  
  118.     {  
  119.         if (result.success() && result.bytes_transferred() > 0) // 发送成功  
  120.         {  
  121.             ACE_Message_Block& mb = result.message_block();  
  122. #ifdef _DEBUG  
  123.             ACE_TCHAR addrStr[128];  
  124.               
  125.             m_ClientAddr.addr_to_string(addrStr, sizeof(addrStr) / sizeof(ACE_TCHAR));  
  126.               
  127.             ACE_DEBUG((LM_INFO, ACE_TEXT("Send to client: %s len:%i/n"), addrStr, result.bytes_transferred()));  
  128.               
  129.             char* ptr = mb.rd_ptr();  
  130.               
  131. #endif  
  132.               
  133.             mb.release();  
  134.         }  
  135.     }  
  136.     void TTcpHandler::initCurDataMB()  
  137.     {  
  138.         ACE_NEW_NORETURN(m_CurDataMB, ACE_Message_Block(TCP_PACK_HEADER_SIZE, TCP_DATA_RECEIVE));  
  139.     }  
  140. } // namespace igame  

 

然后是TTcpNetThread，该类的实现也相当简单：

[cpp] view plain copy
  1. #include <ace/Proactor.h>  
  2. #include "TCPNetThread.h"  
  3. namespace igame  
  4. {  
  5.     int TTcpNetThread::open() { return this->activate(); }  
  6.     int TTcpNetThread::close()  
  7.     {  
  8.         ACE_Proactor::instance()->proactor_end_event_loop(); // 终止ACE_Proactor循环  
  9.         this->wait(); // 等待清理现场  
  10.         return 0;  
  11.     }  
  12.       
  13.     int TTcpNetThread::svc()  
  14.     {  
  15.         ACE_INET_Addr listenAddr(DEF_LISTENING_PORT); // 默认监听地址  
  16.         TTcpAcceptor tcpAcceptor; // 接收器  
  17.         // 设置事件  
  18.         tcpAcceptor.setOnClientConnect(m_OnClientConnect);  
  19.         tcpAcceptor.setOnClientDisconnect(m_OnClientDisconnect);  
  20.         tcpAcceptor.setOnClientValidate(m_OnClientValidate);  
  21.         tcpAcceptor.setOnDataReceive(m_OnDataReceive);  
  22.         tcpAcceptor.setOnDataSendFailed(m_OnDataSendFailed);  
  23.         tcpAcceptor.setOnDataSendSucceeded(m_OnDataSendSucceeded);  
  24.           
  25.         // 演出开始  
  26.         if (tcpAcceptor.open(listenAddr, 0, 1, 5, 1, 0, 0) != 0)  
  27.             ACE_ERROR_RETURN((LM_ERROR, ACE_TEXT("%p/n"), ACE_TEXT("failed to open TcpAcceptor errno=%i/n"), errno), -1);  
  28.         // Proactor的事件循环开始  
  29.         ACE_Proactor::instance()->proactor_run_event_loop();  
  30.         ACE_DEBUG((LM_DEBUG, ACE_TEXT("Network fin/n")));  
  31.         return 0;  
  32.     }  
  33. } // namespace igame  

 

最后，对以上三个类进行聚合，封装，就成了TTcp类，在此之前，先定义消息类型：

[cpp] view plain copy
  1. /** 
  2. * @name TCP的ACE_Message_Block类型定义 @see ACE_Message_Block 
  3. * @{ 
  4. */  
  5. /// @brief TCP数据接收  
  6. #define TCP_DATA_RECEIVE            0x5505  
  7. /// @brief TCP客户端连接  
  8. #define TCP_CLIENT_CONNECT          0x5506  
  9. /// @brief TCP客户端断线  
  10. #define TCP_CLIENT_DISCONNECT       0x5507  
  11. /// @brief TCP数据发送  
  12. #define TCP_DATA_SEND               0x5508  
  13. /// @brief TCP数据发送成功  
  14. #define TCP_DATA_SEND_SUCCEEDED     0x5509  
  15. /// @brief TCP数据发送失败  
  16. #define TCP_DATA_SEND_FAILED        0x550A  
  17. /** 
  18. * @} 
  19. */  
  20. /// 默认监听地址:偶的车牌号  
  21. #define DEF_LISTENING_PORT  777  

现在看看TTcp的实现：

唔，太长了，下一篇吧。

 

 

 

此乃末技，

不知何用。

堆砌字数，

凑成更新。

走过路过，

不要错过。

 请原谅偶拖篇幅，这里奉上拖欠的数字。

TTcp的实现如下：

 

[cpp] view plain copy
  1. #include "Tcp.h"  
  2. namespace igame  
  3. {  
  4.     TTcp::TTcp()  
  5.         :m_TcpNetThd(0)  
  6.     {  
  7.         ACE_NEW_NORETURN(m_TcpNetThd, TTcpNetThread()); // 创建TTcpNetThread对象实例  
  8.     }  
  9.     TTcp::~TTcp()  
  10.     {  
  11.         if (m_TcpNetThd) // 释放  
  12.             delete m_TcpNetThd;  
  13.     }  
  14.     void TTcp::open()  
  15.     {  
  16.         ACE_TRACE("TTcp::open");  
  17.           
  18.         // 所有TTcpNetThread的事件，交由TTcp来处理  
  19.         // TOnClientValidate除外，该事件需要特定的逻辑，且无法异步  
  20.         if (m_TcpNetThd)  
  21.         {  
  22.             m_TcpNetThd->setOnClientConnect(EVENT(TTcpHandler::TOnClientConnect, TTcp, this, tcpNetThread_OnClientConnect));  
  23.             m_TcpNetThd->setOnClientDisconnect(EVENT(TTcpHandler::TOnClientDisconnect, TTcp, this, tcpNetThread_OnClientDisconnect));  
  24.             m_TcpNetThd->setOnClientValidate(m_OnClientValidate);  
  25.             m_TcpNetThd->setOnDataReceive(EVENT(TTcpHandler::TOnDataReceive, TTcp, this, tcpNetThread_OnDataReceive));  
  26.             m_TcpNetThd->setOnDataSendFailed(EVENT(TTcpHandler::TOnDataSendFailed, TTcp, this, tcpNetThread_OnDataSendFailed));  
  27.             m_TcpNetThd->setOnDataSendSucceeded(EVENT(TTcpHandler::TOnDataSendSucceeded, TTcp, this, tcpNetThread_OnDataSendSucceeded));  
  28.         }  
  29.         if (activate() == -1)  
  30.             ACE_DEBUG((LM_ERROR, ACE_TEXT("Resume thread failed")));  
  31.     }  
  32.     void TTcp::close()  
  33.     {  
  34.         if (m_TcpNetThd)  
  35.             m_TcpNetThd->close();  
  36.         ACE_TRACE("TTcp::close");  
  37.         ACE_Message_Block* termBlock; // 结束信号  
  38.           
  39.         ACE_NEW_NORETURN(termBlock, ACE_Message_Block(0, ACE_Message_Block::MB_HANGUP));  
  40.         if (!termBlock)  
  41.             ACE_DEBUG((LM_ERROR, ACE_TEXT("Allocate failed %i"), errno));  
  42.         else  
  43.         {  
  44.             putq(termBlock);  
  45.             wait();  
  46.         }  
  47.     }  
  48.     int TTcp::send(ACE_UINT32 ip, ACE_UINT16 port, unsigned int seq, const char* buf, unsigned short len)  
  49.     {  
  50.         ACE_Message_Block* mb = 0; // 数据包  
  51.         ACE_NEW_RETURN(mb, ACE_Message_Block(sizeof(ACE_UINT32) + sizeof(ACE_UINT16) + sizeof(unsigned int) + sizeof(unsigned short) + len, TCP_DATA_SEND), -1);  
  52.           
  53.         // 格式：ip | port | seq | len | 数据...  
  54.         mb->copy((const char *)&ip, sizeof(ACE_UINT32));  
  55.         mb->copy((const char *)&port, sizeof(ACE_UINT16));  
  56.         mb->copy((const char *)&seq, sizeof(unsigned int));  
  57.         mb->copy((const char *)&len, sizeof(unsigned short));  
  58.         mb->copy(buf, len);  
  59.         return putq(mb);  
  60.     }  
  61.     int TTcp::svc()  
  62.     {  
  63.         ACE_TRACE("TTcp::svc");  
  64.         if (m_TcpNetThd->open() == -1)  
  65.             ACE_DEBUG((LM_ERROR, ACE_TEXT("Failed to pen TTcpNetThread: %i"), errno));  
  66.         ACE_Message_Block* msg = 0;  
  67.         while(true)  
  68.         {  
  69.             if (getq(msg) == -1)  
  70.             {  
  71.                 ACE_ERROR_RETURN((LM_ERROR, ACE_TEXT("Failed to getq %i"), errno), -1);  
  72.             }  
  73.             switch(msg->msg_type())  
  74.             {  
  75.             case ACE_Message_Block::MB_HANGUP: // 偶要退出  
  76.                 {  
  77.                     ACE_DEBUG((LM_DEBUG, ACE_TEXT("Quit")));  
  78.                     msg->release();  
  79.                     return 0;  
  80.                 }  
  81.                 break;  
  82.             case TCP_CLIENT_CONNECT: // 客户端连接  
  83.                 {  
  84.                     int len = msg->length();  
  85.                     int hLen = sizeof(TTcpHandler *);  
  86.                     if (msg->length() != TCP_PACK_HEADER_SIZE + sizeof(TTcpHandler *))  
  87.                         ACE_ERROR_RETURN((LM_ERROR, ACE_TEXT("Tcp connection message block invalid!")), -1);  
  88.                     char* ptr = msg->rd_ptr();  
  89.                     ACE_UINT32 ip = *(ACE_UINT32 *)ptr; ptr += sizeof(ACE_UINT32);  
  90.                     ACE_UINT16 port = *(ACE_UINT16 *)ptr; ptr += sizeof(ACE_UINT16);  
  91.                     TTcpHandler* handler = (TTcpHandler *)(*(int *)ptr);  
  92.                     {  
  93.                         ACE_Guard<ACE_Recursive_Thread_Mutex> lock(m_Lock);  
  94.                         m_AddrMap.insert(make_pair<unsigned __int64, TTcpHandler *>((unsigned __int64)ip << 32 | port, handler));   
  95.                     }                 
  96.                       
  97.                     m_OnClientConnect(ip, port, handler);  
  98.                 }  
  99.                 break;  
  100.             case TCP_CLIENT_DISCONNECT: // 客户端断开连接  
  101.                 {  
  102.                     if (msg->length() != sizeof(ACE_UINT32) + sizeof(ACE_UINT16))  
  103.                         ACE_ERROR_RETURN((LM_ERROR, ACE_TEXT("Invalid tcp disconnect message block/n")), -1);  
  104.                     char* ptr = msg->rd_ptr();  
  105.                     ACE_UINT32 ip = *(ACE_UINT32 *)ptr; ptr += sizeof(ACE_UINT32);  
  106.                     ACE_UINT16 port = *(ACE_UINT16 *)ptr;  
  107.                     {  
  108.                         ACE_Guard<ACE_Recursive_Thread_Mutex> lock(m_Lock);  
  109.                         m_AddrMap.erase((unsigned __int64)ip << 32 | port);  
  110.                     }  
  111.                     m_OnClientDisconnect(ip, port);  
  112.                 }  
  113.                 break;  
  114.             case TCP_DATA_RECEIVE:  
  115.                 {  
  116.                     char* ptr = msg->rd_ptr();  
  117.                     ACE_UINT32 ip = *(ACE_UINT32 *)ptr; ptr += sizeof(ACE_UINT32);  
  118.                     ACE_UINT16 port = *(ACE_UINT16 *)ptr; ptr += sizeof(ACE_UINT16);  
  119.                     TTcpPackHeader* header = (TTcpPackHeader *)ptr; ptr += TCP_PACK_HEADER_SIZE;  
  120.                     const char* data = ptr;  
  121.                     m_OnDataReceive(ip, port, header->seq, data, header->len);  
  122.                 }  
  123.                 break;  
  124.             case TCP_DATA_SEND:  
  125.                 {  
  126.                     if (msg->length() > sizeof(TTcpPackHeader))  
  127.                     {  
  128.                         char* ptr = msg->rd_ptr();  
  129.                         ACE_UINT32 ip = *(ACE_UINT32 *)ptr; ptr += sizeof(ACE_UINT32);  
  130.                         ACE_UINT16 port = *(ACE_UINT16 *)ptr; ptr += sizeof(ACE_UINT16);  
  131.                         unsigned int seq = *(unsigned int *)ptr; ptr += sizeof(unsigned int);  
  132.                         unsigned short len = *(unsigned short *)ptr; ptr += sizeof(unsigned short);  
  133.                         const char* data = ptr;  
  134.                           
  135.                         {  
  136.                             ACE_Guard<ACE_Recursive_Thread_Mutex> _lock(m_Lock);  
  137.                             hash_map<unsigned __int64, TTcpHandler *>::iterator it = m_AddrMap.find((unsigned __int64)ip << 32 | port);  
  138.                             if (it != m_AddrMap.end())  
  139.                             {  
  140.                                 (*it).second->send(seq, data, len);  
  141.                             }  
  142.                         }  
  143.                     }  
  144.                 }  
  145.                 break;  
  146.             case TCP_DATA_SEND_SUCCEEDED:  
  147.                 {  
  148.                     char* ptr = msg->rd_ptr();  
  149.                     ACE_UINT32 ip = *(ACE_UINT32 *)ptr; ptr += sizeof(ACE_UINT32);  
  150.                     ACE_UINT16 port = *(ACE_UINT16 *)ptr; ptr += sizeof(ACE_UINT16);  
  151.                     TTcpPackHeader* header = (TTcpPackHeader *)ptr; ptr += TCP_PACK_HEADER_SIZE;  
  152.                     const char* data = ptr;  
  153.                     m_OnDataSendSucceeded(ip, port, header->seq, data, header->len);  
  154.                 }  
  155.                 break;  
  156.             case TCP_DATA_SEND_FAILED:  
  157.                 {  
  158.                     char* ptr = msg->rd_ptr();  
  159.                     ACE_UINT32 ip = *(ACE_UINT32 *)ptr; ptr += sizeof(ACE_UINT32);  
  160.                     ACE_UINT16 port = *(ACE_UINT16 *)ptr; ptr += sizeof(ACE_UINT16);  
  161.                     TTcpPackHeader* header = (TTcpPackHeader *)ptr; ptr += TCP_PACK_HEADER_SIZE;  
  162.                     const char* data = ptr;  
  163.                     m_OnDataSendFailed(ip, port, header->seq, data, header->len);  
  164.                 }  
  165.                 break;  
  166.             default:  
  167.                 {  
  168.                     ACE_DEBUG((LM_ERROR, ACE_TEXT("Unknown ACE_Message_Block type %i/n"), msg->msg_type()));  
  169.                 }  
  170.                 break;  
  171.             } // switch  
  172.             msg->release();  
  173.         } // while true  
  174.         return 0;  
  175.     }  
  176.     void TTcp::tcpNetThread_OnClientConnect(ACE_UINT32 ip, ACE_UINT16 port, TTcpHandler* handler)  
  177.     {  
  178.         ACE_Message_Block* mb = 0;  
  179.         ACE_NEW_NORETURN(mb, ACE_Message_Block(sizeof(ACE_UINT32) + sizeof(ACE_UINT16) + sizeof(TTcpHandler *), TCP_CLIENT_CONNECT));  
  180.         if (mb)  
  181.         {  
  182.             mb->copy((const char *)&ip, sizeof(ACE_UINT32));  
  183.             mb->copy((const char *)&port, sizeof(ACE_UINT16));  
  184.             mb->copy((const char *)&handler, sizeof(TTcpHandler *));  
  185.             this->putq(mb);  
  186.         }  
  187.     }  
  188.     void TTcp::tcpNetThread_OnClientDisconnect(ACE_UINT32 ip, ACE_UINT16 port)  
  189.     {  
  190.         ACE_Message_Block* mb = 0;  
  191.         ACE_NEW_NORETURN(mb, ACE_Message_Block(sizeof(ACE_UINT32) + sizeof(ACE_UINT16), TCP_CLIENT_DISCONNECT));  
  192.         if (mb)  
  193.         {  
  194.             mb->copy((const char *)&ip, sizeof(ACE_UINT32));  
  195.             mb->copy((const char *)&port, sizeof(ACE_UINT16));  
  196.               
  197.             this->putq(mb);  
  198.         }  
  199.     }  
  200.       
  201.     void TTcp::tcpNetThread_OnDataReceive(ACE_UINT32 ip, ACE_UINT16 port, unsigned int seq, const char* data, unsigned short size)  
  202.     {  
  203.         ACE_Message_Block* mb = 0;  
  204.         ACE_NEW_NORETURN(mb, ACE_Message_Block(sizeof(ACE_UINT32) + sizeof(ACE_UINT16) + TCP_PACK_HEADER_SIZE + size, TCP_DATA_RECEIVE));  
  205.         if (mb)  
  206.         {  
  207.             mb->copy((const char *)&ip, sizeof(ACE_UINT32));  
  208.             mb->copy((const char *)&port, sizeof(ACE_UINT16));  
  209.             mb->copy((const char *)&seq, sizeof(unsigned int));  
  210.             mb->copy((const char *)&size, sizeof(unsigned short));  
  211.             mb->copy(data, size);  
  212.               
  213.             this->putq(mb);  
  214.         }  
  215.     }  
  216.     void TTcp::tcpNetThread_OnDataSendSucceeded(ACE_UINT32 ip, ACE_UINT16 port, unsigned int seq, const char* data, unsigned short size)  
  217.     {  
  218.         ACE_Message_Block* mb = 0;  
  219.         ACE_NEW_NORETURN(mb, ACE_Message_Block(sizeof(ACE_UINT32) + sizeof(ACE_UINT16) + TCP_PACK_HEADER_SIZE + size, TCP_DATA_SEND_SUCCEEDED));  
  220.         if (mb)  
  221.         {  
  222.             mb->copy((const char *)&ip, sizeof(ACE_UINT32));  
  223.             mb->copy((const char *)&port, sizeof(ACE_UINT16));  
  224.             mb->copy((const char *)&seq, sizeof(unsigned int));  
  225.             mb->copy((const char *)&size, sizeof(unsigned short));  
  226.             mb->copy(data, size);  
  227.               
  228.             this->putq(mb);  
  229.         }  
  230.     }  
  231.     void TTcp::tcpNetThread_OnDataSendFailed(ACE_UINT32 ip, ACE_UINT16 port, unsigned int seq, const char* data, unsigned short size)  
  232.     {  
  233.         ACE_Message_Block* mb = 0;  
  234.         ACE_NEW_NORETURN(mb, ACE_Message_Block(sizeof(ACE_UINT32) + sizeof(ACE_UINT16) + TCP_PACK_HEADER_SIZE + size, TCP_DATA_SEND_FAILED));  
  235.         if (mb)  
  236.         {  
  237.             mb->copy((const char *)&ip, sizeof(ACE_UINT32));  
  238.             mb->copy((const char *)&port, sizeof(ACE_UINT16));  
  239.             mb->copy((const char *)&seq, sizeof(unsigned int));  
  240.             mb->copy((const char *)&size, sizeof(unsigned short));  
  241.             mb->copy(data, size);  
  242.               
  243.             this->putq(mb);  
  244.         }  
  245.     }  
  246. } // namespace igame  

 

在完整的工程中，还有测试代码，这里就不列出了。本来已经在下载频道中上传了，并设置下载点数为0，结果传完后楞是自私都找不到？！NNDCSDN！！


这是下载资源。

 

来信到igame2000@hotmail.com

 

需要完整代码的请来信索取吧，必复。

 

 

此乃末技。。。。

 


这篇关于ACE_Proactor网络通信示例的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/463171。

相关文章

Java调用DeepSeek API的最佳实践及详细代码示例

Java调用DeepSeek API的最佳实践及详细代码示例

《Java调用DeepSeekAPI的最佳实践及详细代码示例》：本文主要介绍如何使用Java调用DeepSeekAPI,包括获取API密钥、添加HTTP客户端依赖、创建HTTP请求、处理响应、... 目录1. 获取API密钥2. 添加HTTP客户端依赖3. 创建HTTP请求4. 处理响应5. 错误处理6.
阅读更多...
Android 悬浮窗开发示例((动态权限请求 | 前台服务和通知 | 悬浮窗创建 )

Android 悬浮窗开发示例((动态权限请求 | 前台服务和通知 | 悬浮窗创建 )

《Android悬浮窗开发示例((动态权限请求|前台服务和通知|悬浮窗创建)》本文介绍了Android悬浮窗的实现效果,包括动态权限请求、前台服务和通知的使用,悬浮窗权限需要动态申请并引导... 目录一、悬浮窗 动态权限请求1、动态请求权限2、悬浮窗权限说明3、检查动态权限4、申请动态权限5、权限设置完毕后
阅读更多...
在 Spring Boot 中使用 @Autowired和 @Bean注解的示例详解

在 Spring Boot 中使用 @Autowired和 @Bean注解的示例详解

《在SpringBoot中使用@Autowired和@Bean注解的示例详解》本文通过一个示例演示了如何在SpringBoot中使用@Autowired和@Bean注解进行依赖注入和Bean... 目录在 Spring Boot 中使用 @Autowired 和 @Bean 注解示例背景1. 定义 Stud
阅读更多...
oracle DBMS_SQL.PARSE的使用方法和示例

oracle DBMS_SQL.PARSE的使用方法和示例

《oracleDBMS_SQL.PARSE的使用方法和示例》DBMS_SQL是Oracle数据库中的一个强大包,用于动态构建和执行SQL语句,DBMS_SQL.PARSE过程解析SQL语句或PL/S... 目录语法示例注意事项DBMS_SQL 是 oracle 数据库中的一个强大包，它允许动态地构建和执行
阅读更多...
Python中顺序结构和循环结构示例代码

Python中顺序结构和循环结构示例代码

《Python中顺序结构和循环结构示例代码》：本文主要介绍Python中的条件语句和循环语句,条件语句用于根据条件执行不同的代码块,循环语句用于重复执行一段代码,文章还详细说明了range函数的使... 目录一、条件语句（1）条件语句的定义（2）条件语句的语法（a）单分支 if（b）双分支 if-else（
阅读更多...
Python中Markdown库的使用示例详解

Python中Markdown库的使用示例详解

《Python中Markdown库的使用示例详解》Markdown库是一个用于处理Markdown文本的Python工具,这篇文章主要为大家详细介绍了Markdown库的具体使用,感兴趣的... 目录一、背景二、什么是 Markdown 库三、如何安装这个库四、库函数使用方法1. markdown.mark
阅读更多...
MySQL数据库函数之JSON_EXTRACT示例代码

MySQL数据库函数之JSON_EXTRACT示例代码

《MySQL数据库函数之JSON_EXTRACT示例代码》：本文主要介绍MySQL数据库函数之JSON_EXTRACT的相关资料,JSON_EXTRACT()函数用于从JSON文档中提取值,支持对... 目录前言基本语法路径表达式示例示例 1: 提取简单值示例 2: 提取嵌套值示例 3: 提取数组中的值注意
阅读更多...
CSS3中使用flex和grid实现等高元素布局的示例代码

CSS3中使用flex和grid实现等高元素布局的示例代码

《CSS3中使用flex和grid实现等高元素布局的示例代码》：本文主要介绍了使用CSS3中的Flexbox和Grid布局实现等高元素布局的方法，通过简单的两列实现、每行放置3列以及全部代码的展示，展示了这两种布局方式的实现细节和效果，详细内容请阅读本文，希望能对你有所帮助... 过往的实现方法是使用浮动加
阅读更多...
css渐变色背景|＜gradient示例详解

css渐变色背景|＜gradient示例详解

《css渐变色背景|＜gradient示例详解》CSS渐变是一种从一种颜色平滑过渡到另一种颜色的效果,可以作为元素的背景,它包括线性渐变、径向渐变和锥形渐变，本文介绍css渐变色背景|＜gradien... 使用渐变色作为背景可以直接将渐China编程变色用作元素的背景，可以看做是一种特殊的背景图片。（是作为背
阅读更多...
JAVA调用Deepseek的api完成基本对话简单代码示例

JAVA调用Deepseek的api完成基本对话简单代码示例

《JAVA调用Deepseek的api完成基本对话简单代码示例》：本文主要介绍JAVA调用Deepseek的api完成基本对话的相关资料,文中详细讲解了如何获取DeepSeekAPI密钥、添加H... 获取API密钥首先，从DeepSeek平台获取API密钥，用于身份验证。添加HTTP客户端依赖使用Jav
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2