Boost.Asio C++ 網(wǎng)絡(luò)編程：TCP回顯客戶端/服務(wù)端
[導(dǎo)讀]? ? ? ?回顯就是服務(wù)端將接收到的任何內(nèi)容回發(fā)給客戶端顯示，然后關(guān)閉客戶端的連接。這個(gè)服務(wù)端可以處理任何數(shù)量的客戶端。每個(gè)客戶端連接之后發(fā)送一個(gè)消息，服務(wù)端接收到消息后把它發(fā)送回去。在那之后，服務(wù)
? ? ? ?回顯就是服務(wù)端將接收到的任何內(nèi)容回發(fā)給客戶端顯示，然后關(guān)閉客戶端的連接。這個(gè)服務(wù)端可以處理任何數(shù)量的客戶端。每個(gè)客戶端連接之后發(fā)送一個(gè)消息，服務(wù)端接收到消息后把它發(fā)送回去。在那之后，服務(wù)端關(guān)閉連接。具體流程如下圖所示。
? ? ? ?對(duì)于TCP而言，我們需要一個(gè)額外的保證：每一個(gè)消息以換行符結(jié)束(‘n’)。編寫(xiě)一個(gè)同步回顯服務(wù)端/客戶端非常簡(jiǎn)單。下面我們分別實(shí)現(xiàn)同步客戶端，同步服務(wù)端，異步客戶端和異步服務(wù)端。
一.TCP同步客戶端
#ifdef?WIN32
#define?_WIN32_WINNT?0x0501
#include#endif
#include#include#include#include#includeusing?namespace?boost::asio;
using?boost::system::error_code;
io_service?service;
size_t?read_complete(char?*?buf,?const?error_code?&?err,?size_t?bytes)?{
if?(err)?return?0;
bool?found?=?std::find(buf,?buf?+?bytes,?'n')?<?buf?+?bytes;
//?一個(gè)一個(gè)字符的讀取，直到回車,?不緩存
return?found???0?:?1;
}
ip::tcp::endpoint?ep(ip::address::from_string("127.0.0.1"),?8001);
void?sync_echo(std::string?msg)?{
msg?+=?"n";
ip::tcp::socket?sock(service);
sock.connect(ep);
sock.write_some(buffer(msg));
char?buf[1024];
int?bytes?=?read(sock,?buffer(buf),?boost::bind(read_complete,?buf,?_1,?_2));
std::string?copy(buf,?bytes?-?1);
msg?=?msg.substr(0,?msg.size()?-?1);
std::cout?<<?"server?echoed?our?"?<<?msg?<<?":?"
<<?(copy?==?msg???"OK"?:?"FAIL")?<<?std::endl;
sock.close();
}
int?main(int?argc,?char*?argv[])?{
//?連接多個(gè)客戶端
char*?messages[]?=?{?"Can",?"ge",?"ge",?"blog!",?0?};
boost::thread_group?threads;
for?(char?**?message?=?messages;?*message;?++message)?{
threads.create_thread(boost::bind(sync_echo,?*message));
boost::this_thread::sleep(boost::posix_time::millisec(100));
}
threads.join_all();
system("pause");
}? ? ? ?你會(huì)發(fā)現(xiàn)，在讀取時(shí)，我使用了自由函數(shù)（不屬于socket類，屬于命名空間asio）read()，因?yàn)槲蚁胍x‘n’之前的所有內(nèi)容。sock.read_some()方法滿足不了這個(gè)要求，因?yàn)樗粫?huì)讀可用的，不一定是整個(gè)的消息。
? ? ? ?read(stream, buffer [, completion])這個(gè)方法同步地從一個(gè)流中讀取數(shù)據(jù)。你可以選擇指定一個(gè)完成處理方法。完成處理方法會(huì)在每次read操作調(diào)用成功之后調(diào)用，然后告訴read操作是否完成（如果沒(méi)有完成，它會(huì)繼續(xù)讀取）。它的格式是：size_t completion(const boost::system::error_code& err, size_t bytes_transfered) 。當(dāng)這個(gè)完成處理方法返回0時(shí)，我們認(rèn)為read操作完成；如果它返回一個(gè)非0值，它表示了下一次sock.read_some操作需要從流中讀取的字節(jié)數(shù)。
? ? ? ?read_complete一個(gè)個(gè)的讀取字符，直到回車，這是通過(guò)std::find方法控制的，std::find的行為大概如下。
templateInputIterator?find?(InputIterator?first,?InputIterator?last,?const?T&?val)
{
??while?(first!=last)?{
????if?(*first==val)?return?first;
????++first;
??}
??return?last;
}? ? ? ?結(jié)合到上面客戶端代碼就是，如果沒(méi)有找到回車'n'，std::find始終返回buf+bytes，否則返回'n'的地址，也就是buf+bytes-1，此時(shí)'n'是已讀取內(nèi)容的最后一個(gè)字符。
? ? ? ?注意：因?yàn)槲覀兪峭降?，所以不需要調(diào)用service.run()。
二.TCP同步服務(wù)端
#ifdef?WIN32
#define?_WIN32_WINNT?0x0501
#include#endif
#include#include#include#includeusing?namespace?boost::asio;
using?namespace?boost::posix_time;
using?boost::system::error_code;
io_service?service;
size_t?read_complete(char?*?buff,?const?error_code?&?err,?size_t?bytes)?{
????if?(?err)?return?0;
????bool?found?=?std::find(buff,?buff?+?bytes,?'n')?<?buff?+?bytes;
????//?we?read?one-by-one?until?we?get?to?enter,?no?buffering
????return?found???0?:?1;
}
void?handle_connections()?{
????ip::tcp::acceptor?acceptor(service,?ip::tcp::endpoint(ip::tcp::v4(),8001));
????char?buff[1024];
????while?(?true)?{
????????ip::tcp::socket?sock(service);
????????acceptor.accept(sock);
????????int?bytes?=?read(sock,?buffer(buff),?
????????????????????boost::bind(read_complete,buff,_1,_2));
????????std::string?msg(buff,?bytes);
????????sock.write_some(buffer(msg));
????????sock.close();
????}
}
int?main(int?argc,?char*?argv[])?{
????handle_connections();
}? ? ? ?服務(wù)端的邏輯主要在handle_connections()。因?yàn)槭菃尉€程，它接受一個(gè)客戶端請(qǐng)求，讀取客戶端發(fā)送的消息，然后回發(fā)給客戶端，接著等待下一個(gè)連接?？梢源_定，當(dāng)兩個(gè)客戶端同時(shí)連接時(shí)，第二個(gè)客戶端需要等待服務(wù)端處理完第一個(gè)客戶端的請(qǐng)求。
? ? ? ?還是要注意因?yàn)槲覀兪峭降?，所以不需要調(diào)用service.run()。
? ? ? ?下面是客戶端回顯的結(jié)果，當(dāng)然要先啟動(dòng)服務(wù)端。
三.TCP異步客戶端
#ifdef?WIN32
#define?_WIN32_WINNT?0x0501
#include#endif
#include#include#include#include#includeusing?namespace?boost::asio;
io_service?service;
#define?MEM_FN(x)???????boost::bind(&self_type::x,?shared_from_this())
#define?MEM_FN1(x,y)????boost::bind(&self_type::x,?shared_from_this(),y)
#define?MEM_FN2(x,y,z)??boost::bind(&self_type::x,?shared_from_this(),y,z)
class?talk_to_svr?:?public?boost::enable_shared_from_this
,?boost::noncopyable?{
typedef?talk_to_svr?self_type;
talk_to_svr(const?std::string?&?message)
:?sock_(service),?started_(true),?message_(message)?{}
void?start(ip::tcp::endpoint?ep)?{
sock_.async_connect(ep,?MEM_FN1(on_connect,?_1));
}
public:
typedef?boost::system::error_code?error_code;
typedef?boost::shared_ptrptr;
static?ptr?start(ip::tcp::endpoint?ep,?const?std::string?&?message)?{
ptr?new_(new?talk_to_svr(message));
new_->start(ep);
return?new_;
}
void?stop()?{
if?(!started_)?return;
started_?=?false;
sock_.close();
}
bool?started()?{?return?started_;?}
private:
void?on_connect(const?error_code?&?err)?{
if?(!err)??????do_write(message_?+?"n");
else????????????stop();
}
void?on_read(const?error_code?&?err,?size_t?bytes)?{
if?(!err)?{
std::string?copy(read_buffer_,?bytes?-?1);
std::cout?<<?"server?echoed?our?"?<<?message_?<<?":?"
<<?(copy?==?message_???"OK"?:?"FAIL")?<<?std::endl;
}
stop();
}
void?on_write(const?error_code?&?err,?size_t?bytes)?{
do_read();
}
void?do_read()?{
async_read(sock_,?buffer(read_buffer_),
MEM_FN2(read_complete,?_1,?_2),?MEM_FN2(on_read,?_1,?_2));
}
void?do_write(const?std::string?&?msg)?{
if?(!started())?return;
std::copy(msg.begin(),?msg.end(),?write_buffer_);
sock_.async_write_some(buffer(write_buffer_,?msg.size()),
MEM_FN2(on_write,?_1,?_2));
}
size_t?read_complete(const?boost::system::error_code?&?err,?size_t?bytes)?{
if?(err)?return?0;
bool?found?=?std::find(read_buffer_,?read_buffer_?+?bytes,?'n')?<?read_buffer_?+?bytes;
return?found???0?:?1;
}
private:
ip::tcp::socket?sock_;
enum?{?max_msg?=?1024?};
char?read_buffer_[max_msg];
char?write_buffer_[max_msg];
bool?started_;
std::string?message_;
};
int?main(int?argc,?char*?argv[])?{
ip::tcp::endpoint?ep(ip::address::from_string("127.0.0.1"),?8001);
char*?messages[]?=?{?"Can",?"ge",?"ge",?"blog",?0?};
for?(char?**?message?=?messages;?*message;?++message)?{
talk_to_svr::start(ep,?*message);
boost::this_thread::sleep(boost::posix_time::millisec(100));
}
service.run();
system("pause");
}四.TCP異步服務(wù)端
#ifdef?WIN32
#define?_WIN32_WINNT?0x0501
#include#endif
#include#include#include#includeusing?namespace?boost::asio;
using?namespace?boost::posix_time;
io_service?service;
#define?MEM_FN(x)???????boost::bind(&self_type::x,?shared_from_this())
#define?MEM_FN1(x,y)????boost::bind(&self_type::x,?shared_from_this(),y)
#define?MEM_FN2(x,y,z)??boost::bind(&self_type::x,?shared_from_this(),y,z)
class?talk_to_client?:?public?boost::enable_shared_from_this,?boost::noncopyable?{
typedef?talk_to_client?self_type;
talk_to_client()?:?sock_(service),?started_(false)?{}
public:
typedef?boost::system::error_code?error_code;
typedef?boost::shared_ptrptr;
void?start()?{
started_?=?true;
do_read();
}
static?ptr?new_()?{
ptr?new_(new?talk_to_client);
return?new_;
}
void?stop()?{
if?(!started_)?return;
started_?=?false;
sock_.close();
}
ip::tcp::socket?&?sock()?{?return?sock_;?}
private:
void?on_read(const?error_code?&?err,?size_t?bytes)?{
if?(!err)?{
std::string?msg(read_buffer_,?bytes);
//?echo?message?back,?and?then?stop
do_write(msg?+?"n");
}
stop();
}
void?on_write(const?error_code?&?err,?size_t?bytes)?{
do_read();
}
void?do_read()?{
async_read(sock_,?buffer(read_buffer_),
MEM_FN2(read_complete,?_1,?_2),?MEM_FN2(on_read,?_1,?_2));
}
void?do_write(const?std::string?&?msg)?{
std::copy(msg.begin(),?msg.end(),?write_buffer_);
sock_.async_write_some(buffer(write_buffer_,?msg.size()),
MEM_FN2(on_write,?_1,?_2));
}
size_t?read_complete(const?boost::system::error_code?&?err,?size_t?bytes)?{
if?(err)?return?0;
bool?found?=?std::find(read_buffer_,?read_buffer_?+?bytes,?'n')?<?read_buffer_?+?bytes;
//?we?read?one-by-one?until?we?get?to?enter,?no?buffering
return?found???0?:?1;
}
private:
ip::tcp::socket?sock_;
enum?{?max_msg?=?1024?};
char?read_buffer_[max_msg];
char?write_buffer_[max_msg];
bool?started_;
};
ip::tcp::acceptor?acceptor(service,?ip::tcp::endpoint(ip::tcp::v4(),?8001));
void?handle_accept(talk_to_client::ptr?client,?const?boost::system::error_code?&?err)?{
client->start();
talk_to_client::ptr?new_client?=?talk_to_client::new_();
acceptor.async_accept(new_client->sock(),?boost::bind(handle_accept,?new_client,?_1));
}
int?main(int?argc,?char*?argv[])?{
talk_to_client::ptr?client?=?talk_to_client::new_();
acceptor.async_accept(client->sock(),?boost::bind(handle_accept,?client,?_1));
service.run();
}? ? ? TCP異步客戶端和異步服務(wù)端的關(guān)鍵是enable_shared_from_this模板類的使用，關(guān)于enable_shared_from_this詳見(jiàn)：C++11新特性之十一：enable_shared_from_this，C++11和boost的enable_shared_from_this功能和原理一樣。
? ? ? ?客戶端回顯結(jié)果和同步時(shí)的一樣，如下：
下載文檔
