注：本文僅對(duì)使用ACE進(jìn)行網(wǎng)絡(luò)通信進(jìn)行演示說(shuō)明。本文中的代碼皆使用doxgen的注釋風(fēng)格。本文中使用的事件機(jī)制，其原理與實(shí)現(xiàn)請(qǐng)參考[ 基于C++的事件機(jī)制設(shè)計(jì)［2.0］]一文。
?
ACE的Proactor對(duì)Epoll和IOCP進(jìn)行了良好包裝，因此，使用ACE來(lái)進(jìn)行網(wǎng)絡(luò)開(kāi)發(fā)是相當(dāng)?shù)谋憷?#xff0c;性能也不差。閑言少敘，看代碼。
這里以TCP協(xié)議進(jìn)行流式通信。我們需要解析流，得出每次接收到的數(shù)據(jù)包大小和包含的數(shù)據(jù)域，假定我們的包結(jié)構(gòu)如下：
?
包序列號(hào)(32Bit) | 長(zhǎng)度(16Bit) | 數(shù)據(jù)域(大小為長(zhǎng)度所表示的字節(jié))... | (下一包)
?
通過(guò)分析由包序列號(hào)和長(zhǎng)度組成的包頭來(lái)解決半包，粘包等問(wèn)題，許多其它文章也有描述，這里就省略了。
這樣可以確定我們的包頭結(jié)構(gòu)如下：
?
[cpp] view plain
copy #pragma?pack(push)??#pragma?pack(1)??????????????typedef?struct?tag_TTcpPackHeader??????{??????????unsigned?int?seq;???????????unsigned?short?len;???????}TTcpPackHeader;??#pragma?pack(pop)????#define?TCP_PACK_HEADER_SIZE?sizeof(tag_TTcpPackHeader)??
.
需要注意的是，要求在字節(jié)邊界對(duì)齊。
?
現(xiàn)在來(lái)看看通過(guò)ACE來(lái)實(shí)現(xiàn)TCP通信需要哪些東西：
INET_Addr 用于地址訪問(wèn)
Task_Base 用于線程模型
Message_Block 用于消息傳遞和數(shù)據(jù)容器
Asynch_IO 異步通信
Proactor IOCP架構(gòu)
?
并且，要建立這樣的通信架構(gòu)，我們需要：
一個(gè)Acceptor：用于接受連接
一個(gè)Handler：對(duì)應(yīng)于每個(gè)連接句柄，并用于數(shù)據(jù)的發(fā)送/接收。
一個(gè)事件分發(fā)線程：以事件的形式將接收到數(shù)據(jù)分發(fā)出去，并在對(duì)應(yīng)的句柄上進(jìn)行數(shù)據(jù)發(fā)送。
?
本示例并沒(méi)有采用在接收到數(shù)據(jù)時(shí)立即進(jìn)行處理的方式，而是通過(guò)創(chuàng)建一個(gè)額外的事件分發(fā)線程的形式，將數(shù)據(jù)包投遞到該線程的消息隊(duì)列中，由該線程向外派送。因此，數(shù)據(jù)處理與網(wǎng)絡(luò)層是隔離的，且網(wǎng)絡(luò)層能專注于通信，最大的發(fā)揮效用。
?
好了，下面來(lái)看看實(shí)現(xiàn)：
先看Handler，參考ACE_Service_Handler，我們需要重載open()，addresses(), handle_read_stream()，handle_write_stream()，以在連接打開(kāi)時(shí)進(jìn)行讀寫流對(duì)象的初始化、獲取客戶端地址，處理輸入/輸入流。
?
注：以T作為類的開(kāi)頭而不是C，是出于對(duì)曾經(jīng)偉大的BORLAND的深刻懷念。
注：成員又以m_開(kāi)頭，是出于對(duì)現(xiàn)而今仍偉大的MS的深刻怨念。
?
[cpp] view plain
copy ?????class?TTcpHandler?:?public?ACE_Service_Handler??{??public:?????????????????typedef?TEvent<void,?ACE_UINT32,?ACE_UINT16,?TTcpHandler?*>?TOnClientConnect;????????????????typedef?TEvent<void,?ACE_UINT32,?ACE_UINT16>?TOnClientDisconnect;???????????????????typedef?TEvent<bool,?ACE_UINT32,?ACE_UINT16>?TOnClientValidate;???????????????????typedef?TEvent<void,?ACE_UINT32,?ACE_UINT16,?unsigned?int,?const?char*,?unsigned?short>?TOnDataReceive;?????????????????????????typedef?TEvent<void,?ACE_UINT32,?ACE_UINT16,?unsigned?int,?const?char*,?unsigned?short>?TOnDataSendSucceeded;?????????????????????????typedef?TEvent<void,?ACE_UINT32,?ACE_UINT16,?unsigned?int,?const?char*,?unsigned?short>?TOnDataSendFailed;??private:??????ACE_Asynch_Read_Stream?m_Reader;???????ACE_Asynch_Write_Stream?m_Writer;???????ACE_Message_Block*?m_CurDataMB;???????ACE_INET_Addr?m_ClientAddr;???public:???????????????DECL_PROP(TOnClientConnect,?OnClientConnect)???????DECL_PROP(TOnClientDisconnect,?OnClientDisconnect)???????DECL_PROP(TOnDataReceive,?OnDataReceive)???????DECL_PROP(TOnDataSendSucceeded,?OnDataSendSucceeded)???????DECL_PROP(TOnDataSendFailed,?OnDataSendFailed)???????????public:????????????TTcpHandler();??????????????????~TTcpHandler();??????????????????????????int?send(unsigned?int?seq,?const?char*?data,?unsigned?short?size);?????????????????????virtual?void?open(ACE_HANDLE?h,?ACE_Message_Block&?mb);???????????????virtual?void?addresses?(const?ACE_INET_Addr?&remote_address,????????????????????????????const?ACE_INET_Addr?&local_address);?????????????????????virtual?void?handle_read_stream(const?ACE_Asynch_Read_Stream::Result&?result);???????????????virtual?void?handle_write_stream(const?ACE_Asynch_Write_Stream::Result&?result);??????????????void?initCurDataMB();??};???
?
而相應(yīng)滴，Acceptor在接受連接時(shí)，產(chǎn)生出的Handler應(yīng)該是TTcpHandler類型，其定義如下：
注意，為了將事件句柄與連接句柄（TTcpHandler）掛鉤，這里重載了make_handler()。而重載validate_connection則是為了讓連接驗(yàn)證事件能夠在恰當(dāng)?shù)臅r(shí)機(jī)被激發(fā)。
[cpp] view plain
copy ???????class?TTcpAcceptor?:?public?ACE_Asynch_Acceptor<TTcpHandler>??{??public:????????????????DECL_PROP(TTcpHandler::TOnClientConnect,?OnClientConnect)??????DECL_PROP(TTcpHandler::TOnClientDisconnect,?OnClientDisconnect)??????DECL_PROP(TTcpHandler::TOnClientValidate,?OnClientValidate)??????DECL_PROP(TTcpHandler::TOnDataReceive,?OnDataReceive)??????DECL_PROP(TTcpHandler::TOnDataSendSucceeded,?OnDataSendSucceeded)??????DECL_PROP(TTcpHandler::TOnDataSendFailed,?OnDataSendFailed)??????????protected:????????????????virtual?int?validate_connection?(const?ACE_Asynch_Accept::Result&?result,?????????????????????????????????const?ACE_INET_Addr?&remote,?????????????????????????????????const?ACE_INET_Addr&?local);???????????????virtual?TTcpHandler*?make_handler(void);??};???
?
有了Acceptor和Handler，還需要使之運(yùn)行于Proactor模式下，因此有了以下線程：
[cpp] view plain
copy ???????class?TTcpNetThread?:?public?ACE_Task_Base??{??public:????????????????DECL_PROP(TTcpHandler::TOnClientConnect,?OnClientConnect)??????DECL_PROP(TTcpHandler::TOnClientDisconnect,?OnClientDisconnect)??????DECL_PROP(TTcpHandler::TOnClientValidate,?OnClientValidate)??????DECL_PROP(TTcpHandler::TOnDataReceive,?OnDataReceive)??????DECL_PROP(TTcpHandler::TOnDataSendSucceeded,?OnDataSendSucceeded)??????DECL_PROP(TTcpHandler::TOnDataSendFailed,?OnDataSendFailed)????????????????????int?open();????????????int?close();??protected:????????????virtual?int?svc();??};??
?
最后再看看事件分發(fā)線程，該線程也是對(duì)上述實(shí)現(xiàn)的聚合和封裝，對(duì)外暴露事件和發(fā)送方法：
注意，該類也負(fù)責(zé)響應(yīng)TTcpNetThread所激發(fā)的事件，所以需要派生自TObject。
?
[cpp] view plain
copy ?????class?TTcp?:?public?TObject,?public?ACE_Task<ACE_MT_SYNCH>??{??public:????????????????typedef?TTcpHandler::TOnClientConnect?TOnClientConnect;??????typedef?TTcpHandler::TOnClientDisconnect?TOnClientDisconnect;??????typedef?TTcpHandler::TOnClientValidate?TOnClientValidate;??????typedef?TTcpHandler::TOnDataReceive?TOnDataReceive;??????typedef?TTcpHandler::TOnDataSendSucceeded?TOnDataSendSucceeded;??????typedef?TTcpHandler::TOnDataSendFailed?TOnDataSendFailed;??????????private:???????????????ACE_Recursive_Thread_Mutex?m_Lock;???????hash_map<unsigned?__int64,?TTcpHandler?*>?m_AddrMap;???????TTcpNetThread*?m_TcpNetThd;??????????public:????????????????DECL_PROP(TTcpHandler::TOnClientConnect,?OnClientConnect)??????DECL_PROP(TTcpHandler::TOnClientDisconnect,?OnClientDisconnect)??????DECL_PROP(TTcpHandler::TOnClientValidate,?OnClientValidate)??????DECL_PROP(TTcpHandler::TOnDataReceive,?OnDataReceive)??????DECL_PROP(TTcpHandler::TOnDataSendSucceeded,?OnDataSendSucceeded)??????DECL_PROP(TTcpHandler::TOnDataSendFailed,?OnDataSendFailed)??????????public:????????????TTcp();????????????~TTcp();????????????void?open();????????????void?close();??????????????????int?send(ACE_UINT32?ip,?ACE_UINT16?port,?unsigned?int?seq,?const?char*?buf,?unsigned?short?len);??private:????????????virtual?int?svc();??private:????????????????void?tcpNetThread_OnClientConnect(ACE_UINT32?ip,?ACE_UINT16?port,?TTcpHandler*?handler);??????void?tcpNetThread_OnClientDisconnect(ACE_UINT32?ip,?ACE_UINT16?port);??????void?tcpNetThread_OnDataReceive(ACE_UINT32?ip,?ACE_UINT16?port,?unsigned?int?seq,?const?char*?data,?unsigned?short?size);??????void?tcpNetThread_OnDataSendSucceeded(ACE_UINT32?ip,?ACE_UINT16?port,?unsigned?int?seq,?const?char*?data,?unsigned?short?size);??????void?tcpNetThread_OnDataSendFailed(ACE_UINT32?ip,?ACE_UINT16?port,?unsigned?int?seq,?const?char*?data,?unsigned?short?size);??????????};???
?
現(xiàn)在基本格調(diào)已經(jīng)確定，需要做的是編寫具體實(shí)現(xiàn)代碼了。
?
?
此乃末技。
?
?
應(yīng)用ACE來(lái)作為底層通信的框架，已經(jīng)是許多年前的技術(shù)了，這里純粹是湊字?jǐn)?shù)，騙更新滴。這樣的老東西，確實(shí)是相當(dāng)?shù)淖屓藷o(wú)語(yǔ)。
?現(xiàn)在我們一步步來(lái)看看實(shí)現(xiàn)：
先是TTcpAcceptor，該類僅重載了兩個(gè)方法，如下：
[cpp] view plain
copy #include?"TCPAcceptor.h"??namespace?igame??{??????int?TTcpAcceptor::validate_connection?(const?ACE_Asynch_Accept::Result&?result,?????????????????????????????????????????const?ACE_INET_Addr?&remote,?????????????????????????????????????????const?ACE_INET_Addr&?local)??????{??????????if?(m_OnClientValidate.valid())????????????????????????????return?m_OnClientValidate(remote.get_ip_address(),?remote.get_port_number())???0?:?-1;??????????else??????????????return?0;???????}??????TTcpHandler*?TTcpAcceptor::make_handler(void)??????{??????????TTcpHandler*?handler?=?0;??????????ACE_NEW_RETURN?(handler,?TTcpHandler(),?0);????????????????????handler->setOnClientConnect(m_OnClientConnect);??????????handler->setOnClientDisconnect(m_OnClientDisconnect);??????????handler->setOnDataReceive(m_OnDataReceive);??????????handler->setOnDataSendSucceeded(m_OnDataSendSucceeded);??????????handler->setOnDataSendFailed(m_OnDataSendFailed);??????????return?handler;??????}??}???
?
復(fù)雜的部分在TTcpHandler，該類不僅需要接收數(shù)據(jù)（拼包），也要處理發(fā)送：
?
[cpp] view plain
copy #include?"TcpHandler.h"??namespace?igame??{??????TTcpHandler::TTcpHandler()??????????:m_CurDataMB(0)???????{?}??????TTcpHandler::~TTcpHandler()??????{??????????if?(handle()?!=?ACE_INVALID_HANDLE)??????????{??????????????ACE_OS::closesocket(handle());???#ifdef?_DEBUG????????????????????????????ACE_TCHAR?remoteAddrStr[128];????????????????????????????m_ClientAddr.addr_to_string(remoteAddrStr,?sizeof(remoteAddrStr)?/?sizeof(ACE_TCHAR));??????????????ACE_DEBUG((LM_INFO,?ACE_TEXT("Disconnect?from?%s/n"),?remoteAddrStr));??#endif????????????????????????????m_OnClientDisconnect(m_ClientAddr.get_ip_address(),?m_ClientAddr.get_port_number());??????????????if?(m_CurDataMB)??????????????????m_CurDataMB->release();??????????}??????}??????int?TTcpHandler::send(unsigned?int?seq,?const?char*?data,?unsigned?short?dataSize)??????{??????????ACE_Message_Block*?dataMB?=?0;????????????????????ACE_NEW_NORETURN(dataMB,?ACE_Message_Block(sizeof(unsigned?int)?+?sizeof(unsigned?short)?+?dataSize));????????????????????short?len?=?dataSize;??????????dataMB->copy((const?char?*)&seq,?sizeof(unsigned?int));???????????dataMB->copy((const?char?*)&len,?sizeof(unsigned?short));??????????dataMB->copy((const?char?*)data,?dataSize);????????????????????????int?ret?=?m_Writer.write(*dataMB,?dataMB->length());???????????if?(ret?==?-1)??????????????m_OnDataSendFailed(m_ClientAddr.get_ip_address(),?m_ClientAddr.get_port_number(),?seq,?data,?dataSize);??????????else??????????????m_OnDataSendSucceeded(m_ClientAddr.get_ip_address(),?m_ClientAddr.get_port_number(),?seq,?data,?dataSize);??????????return?ret;??????}??????void?TTcpHandler::addresses?(const?ACE_INET_Addr?&remote_address,????????????????????????????const?ACE_INET_Addr?&local_address)??????{??????????m_ClientAddr?=?remote_address;???????}??????void?TTcpHandler::open(ACE_HANDLE?h,?ACE_Message_Block&?mb)??????{??????????handle(h);?????????????????????if?(m_Reader.open(*this)?==?-1)???????????{??????????????ACE_ERROR((LM_ERROR,?ACE_TEXT("failed?to?open?read?handle?%i/n"),?errno));??????????????delete?this;??????????????return;??????????}??????????if?(m_Writer.open(*this)?==?-1)???????????{??????????????ACE_ERROR((LM_ERROR,?ACE_TEXT("failed?to?open?write?handle?%i/n"),?errno));??????????????delete?this;??????????????return;??????????}??????????????????????????????m_OnClientConnect(m_ClientAddr.get_ip_address(),?m_ClientAddr.get_port_number(),?this);??????????initCurDataMB();????????????????????m_Reader.read(*m_CurDataMB,?m_CurDataMB->space());???????}??????void?TTcpHandler::handle_read_stream(const?ACE_Asynch_Read_Stream::Result&?result)??????{??????????ACE_Message_Block&?mb?=?result.message_block();??????????if?(!result.success()?||?result.bytes_transferred()?==?0)???????????{??????????????mb.release();??????????????delete?this;??????????}??????????else??????????{??????????????if?(this->m_CurDataMB->length()?<?TCP_PACK_HEADER_SIZE)???????????????{??????????????????this->m_Reader.read(*m_CurDataMB,?m_CurDataMB->space());??????????????????return?;??????????????}??????????????TTcpPackHeader*?header?=?reinterpret_cast<TTcpPackHeader?*>(this->m_CurDataMB->rd_ptr());??????????????ACE_Message_Block*?dataMB?=?this->m_CurDataMB->cont();??????????????if?(!dataMB)??????????????{??????????????????ACE_NEW_NORETURN(dataMB,?ACE_Message_Block(header->len));??????????????????if?(dataMB)??????????????????????this->m_CurDataMB->cont(dataMB);??????????????????else??????????????????{??????????????????????this->m_CurDataMB->release();??????????????????????ACE_DEBUG((LM_ERROR,?ACE_TEXT("Failed?to?allocated:?%i/n"),?errno));??????????????????????delete?this;??????????????????????return?;??????????????????}??????????????}????????????????????????????if?(dataMB->length()?==?header->len)??????????????{????????????????????????????????????m_OnDataReceive(m_ClientAddr.get_ip_address(),?m_ClientAddr.get_port_number(),?header->seq,?dataMB->rd_ptr(),?header->len);????????????????????????????????????m_CurDataMB->release();??????????????????initCurDataMB();?????????????????????????????????????this->m_Reader.read(*m_CurDataMB,?m_CurDataMB->space());???????????????????return?;??????????????}????????????????????????????this->m_Reader.read(*dataMB,?dataMB->space());???????????}??????}??????void?TTcpHandler::handle_write_stream(const?ACE_Asynch_Write_Stream::Result&?result)??????{??????????if?(result.success()?&&?result.bytes_transferred()?>?0)???????????{??????????????ACE_Message_Block&?mb?=?result.message_block();??#ifdef?_DEBUG??????????????ACE_TCHAR?addrStr[128];????????????????????????????m_ClientAddr.addr_to_string(addrStr,?sizeof(addrStr)?/?sizeof(ACE_TCHAR));????????????????????????????ACE_DEBUG((LM_INFO,?ACE_TEXT("Send?to?client:?%s?len:%i/n"),?addrStr,?result.bytes_transferred()));????????????????????????????char*?ptr?=?mb.rd_ptr();????????????????#endif????????????????????????????mb.release();??????????}??????}??????void?TTcpHandler::initCurDataMB()??????{??????????ACE_NEW_NORETURN(m_CurDataMB,?ACE_Message_Block(TCP_PACK_HEADER_SIZE,?TCP_DATA_RECEIVE));??????}??}???
?
然后是TTcpNetThread，該類的實(shí)現(xiàn)也相當(dāng)簡(jiǎn)單：
[cpp] view plain
copy #include?<ace/Proactor.h>??#include?"TCPNetThread.h"??namespace?igame??{??????int?TTcpNetThread::open()?{?return?this->activate();?}??????int?TTcpNetThread::close()??????{??????????ACE_Proactor::instance()->proactor_end_event_loop();???????????this->wait();???????????return?0;??????}????????????int?TTcpNetThread::svc()??????{??????????ACE_INET_Addr?listenAddr(DEF_LISTENING_PORT);???????????TTcpAcceptor?tcpAcceptor;?????????????????????tcpAcceptor.setOnClientConnect(m_OnClientConnect);??????????tcpAcceptor.setOnClientDisconnect(m_OnClientDisconnect);??????????tcpAcceptor.setOnClientValidate(m_OnClientValidate);??????????tcpAcceptor.setOnDataReceive(m_OnDataReceive);??????????tcpAcceptor.setOnDataSendFailed(m_OnDataSendFailed);??????????tcpAcceptor.setOnDataSendSucceeded(m_OnDataSendSucceeded);??????????????????????????????if?(tcpAcceptor.open(listenAddr,?0,?1,?5,?1,?0,?0)?!=?0)??????????????ACE_ERROR_RETURN((LM_ERROR,?ACE_TEXT("%p/n"),?ACE_TEXT("failed?to?open?TcpAcceptor?errno=%i/n"),?errno),?-1);????????????????????ACE_Proactor::instance()->proactor_run_event_loop();??????????ACE_DEBUG((LM_DEBUG,?ACE_TEXT("Network?fin/n")));??????????return?0;??????}??}???
?
最后，對(duì)以上三個(gè)類進(jìn)行聚合，封裝，就成了TTcp類，在此之前，先定義消息類型：
[cpp] view plain
copy ???????#define?TCP_DATA_RECEIVE????????????0x5505????#define?TCP_CLIENT_CONNECT??????????0x5506????#define?TCP_CLIENT_DISCONNECT???????0x5507????#define?TCP_DATA_SEND???????????????0x5508????#define?TCP_DATA_SEND_SUCCEEDED?????0x5509????#define?TCP_DATA_SEND_FAILED????????0x550A????????#define?DEF_LISTENING_PORT??777??
現(xiàn)在看看TTcp的實(shí)現(xiàn)：
唔，太長(zhǎng)了，下一篇吧。
?
?
?
此乃末技，
不知何用。
堆砌字?jǐn)?shù)，
湊成更新。
走過(guò)路過(guò)，
不要錯(cuò)過(guò)。
?請(qǐng)?jiān)徟纪掀?#xff0c;這里奉上拖欠的數(shù)字。
TTcp的實(shí)現(xiàn)如下：
?
[cpp] view plain
copy #include?"Tcp.h"??namespace?igame??{??????TTcp::TTcp()??????????:m_TcpNetThd(0)??????{??????????ACE_NEW_NORETURN(m_TcpNetThd,?TTcpNetThread());???????}??????TTcp::~TTcp()??????{??????????if?(m_TcpNetThd)???????????????delete?m_TcpNetThd;??????}??????void?TTcp::open()??????{??????????ACE_TRACE("TTcp::open");????????????????????????????????????????if?(m_TcpNetThd)??????????{??????????????m_TcpNetThd->setOnClientConnect(EVENT(TTcpHandler::TOnClientConnect,?TTcp,?this,?tcpNetThread_OnClientConnect));??????????????m_TcpNetThd->setOnClientDisconnect(EVENT(TTcpHandler::TOnClientDisconnect,?TTcp,?this,?tcpNetThread_OnClientDisconnect));??????????????m_TcpNetThd->setOnClientValidate(m_OnClientValidate);??????????????m_TcpNetThd->setOnDataReceive(EVENT(TTcpHandler::TOnDataReceive,?TTcp,?this,?tcpNetThread_OnDataReceive));??????????????m_TcpNetThd->setOnDataSendFailed(EVENT(TTcpHandler::TOnDataSendFailed,?TTcp,?this,?tcpNetThread_OnDataSendFailed));??????????????m_TcpNetThd->setOnDataSendSucceeded(EVENT(TTcpHandler::TOnDataSendSucceeded,?TTcp,?this,?tcpNetThread_OnDataSendSucceeded));??????????}??????????if?(activate()?==?-1)??????????????ACE_DEBUG((LM_ERROR,?ACE_TEXT("Resume?thread?failed")));??????}??????void?TTcp::close()??????{??????????if?(m_TcpNetThd)??????????????m_TcpNetThd->close();??????????ACE_TRACE("TTcp::close");??????????ACE_Message_Block*?termBlock;?????????????????????ACE_NEW_NORETURN(termBlock,?ACE_Message_Block(0,?ACE_Message_Block::MB_HANGUP));??????????if?(!termBlock)??????????????ACE_DEBUG((LM_ERROR,?ACE_TEXT("Allocate?failed?%i"),?errno));??????????else??????????{??????????????putq(termBlock);??????????????wait();??????????}??????}??????int?TTcp::send(ACE_UINT32?ip,?ACE_UINT16?port,?unsigned?int?seq,?const?char*?buf,?unsigned?short?len)??????{??????????ACE_Message_Block*?mb?=?0;???????????ACE_NEW_RETURN(mb,?ACE_Message_Block(sizeof(ACE_UINT32)?+?sizeof(ACE_UINT16)?+?sizeof(unsigned?int)?+?sizeof(unsigned?short)?+?len,?TCP_DATA_SEND),?-1);??????????????????????????????mb->copy((const?char?*)&ip,?sizeof(ACE_UINT32));??????????mb->copy((const?char?*)&port,?sizeof(ACE_UINT16));??????????mb->copy((const?char?*)&seq,?sizeof(unsigned?int));??????????mb->copy((const?char?*)&len,?sizeof(unsigned?short));??????????mb->copy(buf,?len);??????????return?putq(mb);??????}??????int?TTcp::svc()??????{??????????ACE_TRACE("TTcp::svc");??????????if?(m_TcpNetThd->open()?==?-1)??????????????ACE_DEBUG((LM_ERROR,?ACE_TEXT("Failed?to?pen?TTcpNetThread:?%i"),?errno));??????????ACE_Message_Block*?msg?=?0;??????????while(true)??????????{??????????????if?(getq(msg)?==?-1)??????????????{??????????????????ACE_ERROR_RETURN((LM_ERROR,?ACE_TEXT("Failed?to?getq?%i"),?errno),?-1);??????????????}??????????????switch(msg->msg_type())??????????????{??????????????case?ACE_Message_Block::MB_HANGUP:???????????????????{??????????????????????ACE_DEBUG((LM_DEBUG,?ACE_TEXT("Quit")));??????????????????????msg->release();??????????????????????return?0;??????????????????}??????????????????break;??????????????case?TCP_CLIENT_CONNECT:???????????????????{??????????????????????int?len?=?msg->length();??????????????????????int?hLen?=?sizeof(TTcpHandler?*);??????????????????????if?(msg->length()?!=?TCP_PACK_HEADER_SIZE?+?sizeof(TTcpHandler?*))??????????????????????????ACE_ERROR_RETURN((LM_ERROR,?ACE_TEXT("Tcp?connection?message?block?invalid!")),?-1);??????????????????????char*?ptr?=?msg->rd_ptr();??????????????????????ACE_UINT32?ip?=?*(ACE_UINT32?*)ptr;?ptr?+=?sizeof(ACE_UINT32);??????????????????????ACE_UINT16?port?=?*(ACE_UINT16?*)ptr;?ptr?+=?sizeof(ACE_UINT16);??????????????????????TTcpHandler*?handler?=?(TTcpHandler?*)(*(int?*)ptr);??????????????????????{??????????????????????????ACE_Guard<ACE_Recursive_Thread_Mutex>?lock(m_Lock);??????????????????????????m_AddrMap.insert(make_pair<unsigned?__int64,?TTcpHandler?*>((unsigned?__int64)ip?<<?32?|?port,?handler));???????????????????????}???????????????????????????????????????????????????????????m_OnClientConnect(ip,?port,?handler);??????????????????}??????????????????break;??????????????case?TCP_CLIENT_DISCONNECT:???????????????????{??????????????????????if?(msg->length()?!=?sizeof(ACE_UINT32)?+?sizeof(ACE_UINT16))??????????????????????????ACE_ERROR_RETURN((LM_ERROR,?ACE_TEXT("Invalid?tcp?disconnect?message?block/n")),?-1);??????????????????????char*?ptr?=?msg->rd_ptr();??????????????????????ACE_UINT32?ip?=?*(ACE_UINT32?*)ptr;?ptr?+=?sizeof(ACE_UINT32);??????????????????????ACE_UINT16?port?=?*(ACE_UINT16?*)ptr;??????????????????????{??????????????????????????ACE_Guard<ACE_Recursive_Thread_Mutex>?lock(m_Lock);??????????????????????????m_AddrMap.erase((unsigned?__int64)ip?<<?32?|?port);??????????????????????}??????????????????????m_OnClientDisconnect(ip,?port);??????????????????}??????????????????break;??????????????case?TCP_DATA_RECEIVE:??????????????????{??????????????????????char*?ptr?=?msg->rd_ptr();??????????????????????ACE_UINT32?ip?=?*(ACE_UINT32?*)ptr;?ptr?+=?sizeof(ACE_UINT32);??????????????????????ACE_UINT16?port?=?*(ACE_UINT16?*)ptr;?ptr?+=?sizeof(ACE_UINT16);??????????????????????TTcpPackHeader*?header?=?(TTcpPackHeader?*)ptr;?ptr?+=?TCP_PACK_HEADER_SIZE;??????????????????????const?char*?data?=?ptr;??????????????????????m_OnDataReceive(ip,?port,?header->seq,?data,?header->len);??????????????????}??????????????????break;??????????????case?TCP_DATA_SEND:??????????????????{??????????????????????if?(msg->length()?>?sizeof(TTcpPackHeader))??????????????????????{??????????????????????????char*?ptr?=?msg->rd_ptr();??????????????????????????ACE_UINT32?ip?=?*(ACE_UINT32?*)ptr;?ptr?+=?sizeof(ACE_UINT32);??????????????????????????ACE_UINT16?port?=?*(ACE_UINT16?*)ptr;?ptr?+=?sizeof(ACE_UINT16);??????????????????????????unsigned?int?seq?=?*(unsigned?int?*)ptr;?ptr?+=?sizeof(unsigned?int);??????????????????????????unsigned?short?len?=?*(unsigned?short?*)ptr;?ptr?+=?sizeof(unsigned?short);??????????????????????????const?char*?data?=?ptr;????????????????????????????????????????????????????{??????????????????????????????ACE_Guard<ACE_Recursive_Thread_Mutex>?_lock(m_Lock);??????????????????????????????hash_map<unsigned?__int64,?TTcpHandler?*>::iterator?it?=?m_AddrMap.find((unsigned?__int64)ip?<<?32?|?port);??????????????????????????????if?(it?!=?m_AddrMap.end())??????????????????????????????{??????????????????????????????????(*it).second->send(seq,?data,?len);??????????????????????????????}??????????????????????????}??????????????????????}??????????????????}??????????????????break;??????????????case?TCP_DATA_SEND_SUCCEEDED:??????????????????{??????????????????????char*?ptr?=?msg->rd_ptr();??????????????????????ACE_UINT32?ip?=?*(ACE_UINT32?*)ptr;?ptr?+=?sizeof(ACE_UINT32);??????????????????????ACE_UINT16?port?=?*(ACE_UINT16?*)ptr;?ptr?+=?sizeof(ACE_UINT16);??????????????????????TTcpPackHeader*?header?=?(TTcpPackHeader?*)ptr;?ptr?+=?TCP_PACK_HEADER_SIZE;??????????????????????const?char*?data?=?ptr;??????????????????????m_OnDataSendSucceeded(ip,?port,?header->seq,?data,?header->len);??????????????????}??????????????????break;??????????????case?TCP_DATA_SEND_FAILED:??????????????????{??????????????????????char*?ptr?=?msg->rd_ptr();??????????????????????ACE_UINT32?ip?=?*(ACE_UINT32?*)ptr;?ptr?+=?sizeof(ACE_UINT32);??????????????????????ACE_UINT16?port?=?*(ACE_UINT16?*)ptr;?ptr?+=?sizeof(ACE_UINT16);??????????????????????TTcpPackHeader*?header?=?(TTcpPackHeader?*)ptr;?ptr?+=?TCP_PACK_HEADER_SIZE;??????????????????????const?char*?data?=?ptr;??????????????????????m_OnDataSendFailed(ip,?port,?header->seq,?data,?header->len);??????????????????}??????????????????break;??????????????default:??????????????????{??????????????????????ACE_DEBUG((LM_ERROR,?ACE_TEXT("Unknown?ACE_Message_Block?type?%i/n"),?msg->msg_type()));??????????????????}??????????????????break;??????????????}???????????????msg->release();??????????}???????????return?0;??????}??????void?TTcp::tcpNetThread_OnClientConnect(ACE_UINT32?ip,?ACE_UINT16?port,?TTcpHandler*?handler)??????{??????????ACE_Message_Block*?mb?=?0;??????????ACE_NEW_NORETURN(mb,?ACE_Message_Block(sizeof(ACE_UINT32)?+?sizeof(ACE_UINT16)?+?sizeof(TTcpHandler?*),?TCP_CLIENT_CONNECT));??????????if?(mb)??????????{??????????????mb->copy((const?char?*)&ip,?sizeof(ACE_UINT32));??????????????mb->copy((const?char?*)&port,?sizeof(ACE_UINT16));??????????????mb->copy((const?char?*)&handler,?sizeof(TTcpHandler?*));??????????????this->putq(mb);??????????}??????}??????void?TTcp::tcpNetThread_OnClientDisconnect(ACE_UINT32?ip,?ACE_UINT16?port)??????{??????????ACE_Message_Block*?mb?=?0;??????????ACE_NEW_NORETURN(mb,?ACE_Message_Block(sizeof(ACE_UINT32)?+?sizeof(ACE_UINT16),?TCP_CLIENT_DISCONNECT));??????????if?(mb)??????????{??????????????mb->copy((const?char?*)&ip,?sizeof(ACE_UINT32));??????????????mb->copy((const?char?*)&port,?sizeof(ACE_UINT16));????????????????????????????this->putq(mb);??????????}??????}????????????void?TTcp::tcpNetThread_OnDataReceive(ACE_UINT32?ip,?ACE_UINT16?port,?unsigned?int?seq,?const?char*?data,?unsigned?short?size)??????{??????????ACE_Message_Block*?mb?=?0;??????????ACE_NEW_NORETURN(mb,?ACE_Message_Block(sizeof(ACE_UINT32)?+?sizeof(ACE_UINT16)?+?TCP_PACK_HEADER_SIZE?+?size,?TCP_DATA_RECEIVE));??????????if?(mb)??????????{??????????????mb->copy((const?char?*)&ip,?sizeof(ACE_UINT32));??????????????mb->copy((const?char?*)&port,?sizeof(ACE_UINT16));??????????????mb->copy((const?char?*)&seq,?sizeof(unsigned?int));??????????????mb->copy((const?char?*)&size,?sizeof(unsigned?short));??????????????mb->copy(data,?size);????????????????????????????this->putq(mb);??????????}??????}??????void?TTcp::tcpNetThread_OnDataSendSucceeded(ACE_UINT32?ip,?ACE_UINT16?port,?unsigned?int?seq,?const?char*?data,?unsigned?short?size)??????{??????????ACE_Message_Block*?mb?=?0;??????????ACE_NEW_NORETURN(mb,?ACE_Message_Block(sizeof(ACE_UINT32)?+?sizeof(ACE_UINT16)?+?TCP_PACK_HEADER_SIZE?+?size,?TCP_DATA_SEND_SUCCEEDED));??????????if?(mb)??????????{??????????????mb->copy((const?char?*)&ip,?sizeof(ACE_UINT32));??????????????mb->copy((const?char?*)&port,?sizeof(ACE_UINT16));??????????????mb->copy((const?char?*)&seq,?sizeof(unsigned?int));??????????????mb->copy((const?char?*)&size,?sizeof(unsigned?short));??????????????mb->copy(data,?size);????????????????????????????this->putq(mb);??????????}??????}??????void?TTcp::tcpNetThread_OnDataSendFailed(ACE_UINT32?ip,?ACE_UINT16?port,?unsigned?int?seq,?const?char*?data,?unsigned?short?size)??????{??????????ACE_Message_Block*?mb?=?0;??????????ACE_NEW_NORETURN(mb,?ACE_Message_Block(sizeof(ACE_UINT32)?+?sizeof(ACE_UINT16)?+?TCP_PACK_HEADER_SIZE?+?size,?TCP_DATA_SEND_FAILED));??????????if?(mb)??????????{??????????????mb->copy((const?char?*)&ip,?sizeof(ACE_UINT32));??????????????mb->copy((const?char?*)&port,?sizeof(ACE_UINT16));??????????????mb->copy((const?char?*)&seq,?sizeof(unsigned?int));??????????????mb->copy((const?char?*)&size,?sizeof(unsigned?short));??????????????mb->copy(data,?size);????????????????????????????this->putq(mb);??????????}??????}??}???
?
在完整的工程中，還有測(cè)試代碼，這里就不列出了。本來(lái)已經(jīng)在下載頻道中上傳了，并設(shè)置下載點(diǎn)數(shù)為0，結(jié)果傳完后楞是自私都找不到？！NNDCSDN！！
這是下載資源。
?
來(lái)信到igame2000@hotmail.com
?
需要完整代碼的請(qǐng)來(lái)信索取吧，必復(fù)。
?
?
此乃末技。。。。
?
《新程序員》：云原生和全面數(shù)字化實(shí)踐50位技術(shù)專家共同創(chuàng)作，文字、視頻、音頻交互閱讀
總結(jié)
以上是生活随笔為你收集整理的ACE_Proactor网络通信示例的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。
如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。
