客户端:

// Client.cpp : 定义控制台应用程序的入口点。
//#include "stdafx.h"
#include <iostream>
#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/enable_shared_from_this.hpp>using boost::asio::ip::tcp;
using boost::asio::ip::address;class session: public boost::enable_shared_from_this<session> {
public:session(boost::asio::io_service &io_service, tcp::endpoint &endpoint): io_service_(io_service), socket_(io_service), endpoint_(endpoint){ } void start() {socket_.async_connect(endpoint_,boost::bind(&session::handle_connect, shared_from_this(), boost::asio::placeholders::error));} private:void handle_connect(const boost::system::error_code &error) {if (error) {if (error.value() != boost::system::errc::operation_canceled) {std::cerr << boost::system::system_error(error).what() << std::endl;} socket_.close();return;} static tcp::no_delay option(true);socket_.set_option(option);strcpy(buf, "Hello World!\n");boost::asio::async_write(socket_, boost::asio::buffer(buf, strlen(buf)),boost::bind(&session::handle_write, shared_from_this(), boost::asio::placeholders::error,boost::asio::placeholders::bytes_transferred));} void handle_write(const boost::system::error_code& error, size_t bytes_transferred) {memset(buf, sizeof(buf), 0); boost::asio::async_read_until(socket_, sbuf,"\n",boost::bind(&session::handle_read,shared_from_this(),boost::asio::placeholders::error,boost::asio::placeholders::bytes_transferred));}void handle_read(const boost::system::error_code& error, size_t bytes_transferred) {std::cout << buf << std::endl;}private:boost::asio::io_service &io_service_;tcp::socket socket_;tcp::endpoint &endpoint_;char buf[1024];boost::asio::streambuf sbuf;
};typedef boost::shared_ptr<session> session_ptr;int main(int argc, char* argv[])
{boost::asio::io_service io_service;tcp::endpoint endpoint(address::from_string("172.16.6.70"), 10028);session_ptr new_session(new session(io_service, endpoint));new_session->start();io_service.run();return 0;
}

服务器:

#include <string.h>
#include <iostream>
#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/enable_shared_from_this.hpp>using boost::asio::ip::tcp;
using boost::asio::ip::address;class session: public boost::enable_shared_from_this<session> {
public:session(boost::asio::io_service &io_service): socket_(io_service){ } void start() {static tcp::no_delay option(true);socket_.set_option(option);boost::asio::async_read_until(socket_, sbuf_,"\n",boost::bind(&session::handle_read, shared_from_this(), boost::asio::placeholders::error,boost::asio::placeholders::bytes_transferred));} tcp::socket &socket() {return socket_;} private:void handle_write(const boost::system::error_code& error, size_t bytes_transferred) {boost::asio::async_read_until(socket_, sbuf_,"\n",boost::bind(&session::handle_read, shared_from_this(), boost::asio::placeholders::error,boost::asio::placeholders::bytes_transferred));} void handle_read(const boost::system::error_code& error, size_t bytes_transferred) {boost::asio::async_write(socket_, sbuf_,boost::bind(&session::handle_write, shared_from_this(), boost::asio::placeholders::error,boost::asio::placeholders::bytes_transferred));std::ostream os(&sbuf_);printf("%s \n",sbuf_.data());}private:tcp::socket socket_;boost::asio::streambuf sbuf_;
};typedef boost::shared_ptr<session> session_ptr;class server {
public:server(boost::asio::io_service &io_service, tcp::endpoint &endpoint): io_service_(io_service), acceptor_(io_service, endpoint){session_ptr new_session(new session(io_service_));acceptor_.async_accept(new_session->socket(),boost::bind(&server::handle_accept,this,new_session,boost::asio::placeholders::error));}void handle_accept(session_ptr new_session, const boost::system::error_code& error) {if (error) {return;}new_session->start();new_session.reset(new session(io_service_));acceptor_.async_accept(new_session->socket(),boost::bind(&server::handle_accept,this,new_session,boost::asio::placeholders::error));}void run() {io_service_.run();}private:boost::asio::io_service &io_service_;tcp::acceptor acceptor_;
};int main(int argc, char* argv[])
{boost::asio::io_service io_service;//tcp::endpoint endpoint(address::from_string("172.16.6.70"), 10028);tcp::endpoint endpoint(boost::asio::ip::tcp::v4(), 10028);server s(io_service, endpoint);s.run();return 0;
}

编译: 
g++ -Wall -o client client.cpp -lboost_system
g++ -Wall -o server server.cpp -lboost_system

这里需要注意的是: async_write, async_read, async_read_until 都是需要达到特定条件才会调用回调函数, 
在调用回调函数之前, 不能再次调用, 否则接收到的数据很可能是乱的. 所以, 在实际代码当中, 会有一个写缓冲队列, 当需要write的时, 先放到队列中, 如果队列个数为1, 则调用async_write, 否则等待函数回调, 当函数回调时将首个元素从队列中移除, 然后接着发送下一个, 直到队列为空.

对于client, 由于is_open在async_connect之后就是true状态了, 因此在async_connect回调返回之前没有方法知道是否已经连接成功, 实际代码当中一般会增加一个变量以表示该套接字是否已经允许发送数据.

============================================================================

服务器:

[cpp] view plain copy  
  1. #define         PACK_MAX_SIZE          16*1024
  2. class TCPConnection: public boost::enable_shared_from_this<TCPConnection> {
  3. public:
  4. static boost::shared_ptr<TCPConnection> create(IoService& ioService);
  5. tcp::socket& getSocket();
  6. void start();
  7. private:
  8. TCPConnection(IoService& ioService);
  9. void handleRead(const boost::system::error_code& e,size_t bytesTransferred);
  10. void handleWrite(const boost::system::error_code& e,size_t bytesTransferred);
  11. //TCP的socket
  12. tcp::socket socket;
  13. //接收和发送的缓冲区
  14. char m_buffer[PACK_MAX_SIZE];
  15. size_t m_bytesReceived;
  16. };
  17. typedef boost::shared_ptr<TCPConnection> pointer;
  18. class TCPServer {
  19. public:
  20. TCPServer(IoService& ioService, int port);
  21. public:
  22. private:
  23. void startAccept();
  24. void handleAccept(boost::shared_ptr<TCPConnection> newConnection,
  25. const boost::system::error_code& error);
  26. private:
  27. tcp::acceptor acceptor;
  28. };
[cpp] view plain copy  
  1. TCPConnection::TCPConnection(IoService& ioService) :socket(ioService) {
  2. m_bytesReceived = 0;
  3. }
  4. boost::shared_ptr<TCPConnection> TCPConnection::create(IoService& ioService) {
  5. return pointer(new TCPConnection(ioService));
  6. }
  7. tcp::socket& TCPConnection::getSocket() {
  8. return socket;
  9. }
  10. void TCPConnection::start() {
  11. static tcp::no_delay option( true );
  12. socket.set_option( option );
  13. socket.set_option(boost::asio::socket_base::keep_alive(true));
  14. memset(m_buffer, 0x0, 2048);
  15. socket.async_read_some(boost::asio::buffer(m_buffer),
  16. boost::bind(&TCPConnection::handleRead, shared_from_this(),
  17. boost::asio::placeholders::error,
  18. boost::asio::placeholders::bytes_transferred));
  19. }
  20. void TCPConnection::sendTermMessage( long phoneNum, unsigned short serialNum, unsigned char cmd, void* rawData )
  21. {
  22. socket.async_write_some(
  23. boost::asio::buffer( resp->mem, resp->size ),
  24. boost::bind(
  25. &TCPConnection::handleWrite,
  26. shared_from_this(),
  27. boost::asio::placeholders::error,
  28. boost::asio::placeholders::bytes_transferred
  29. )
  30. );
  31. }
  32. void TCPConnection::handleRead(const boost::system::error_code& error, size_t bytesTransferred)
  33. {
  34. //cout <<"************************socket read " << bytesTransferred << " bytes codes ***************************"<< endl;
  35. if( error != 0 || bytesTransferred == 0 )
  36. {
  37. std::cout << "!!! network exception: err(" << error.value() << "), " << boost::system::system_error( error ).what() << " !!!" << std::endl;
  38. socket.close();
  39. CmdQueue::getInstance()->cancelSubscribe( this );
  40. return;
  41. }
  42. std::cout << "received some from remote serivce";
  43. time_t nowTime = time(NULL);
  44. std::cout << ":(time:"<<nowTime<<")" << SQ::Utils::bytes2HexString( (SQ::Byte *) m_buffer, bytesTransferred<=40?bytesTransferred:40);
  45. if (bytesTransferred<=40)
  46. {
  47. std::cout<< std::endl;
  48. }
  49. else
  50. {
  51. cout << "..." <<endl;
  52. }
  53. //解析消息（主要考虑到TCP分包和TCP粘包）
  54. size_t m_bytesReceived = 0;
  55. m_bytesReceived += bytesTransferred;
  56. size_t dealDataSize = 0;
  57. if (m_bytesReceived>=sizeof(TCPHead))
  58. {
  59. int offset = 0;
  60. while (true)//考虑到TCP粘包问题，做个循环
  61. {
  62. //剩余缓冲区不足消息头
  63. offset = dealDataSize;
  64. if (m_bytesReceived-offset < sizeof(TCPHead)) break;
  65. TCPHead* head = (TCPHead*)(m_buffer + offset);
  66. unsigned long phoneNum = SQ::Utils::BCD2long( head->terminalID, PHONENUM_LENGTH );
  67. offset += PHONENUM_LENGTH;
  68. unsigned short serialNum = ntohs( head->serialNum );
  69. offset += 2;
  70. unsigned char cmd = head->cmd ;
  71. offset += 1;
  72. int bodyLen = ntohl( head->bodyLen );
  73. offset += 4;
  74. if (m_bytesReceived-offset < bodyLen) break;
  75. SQ::Byte *pRawData = (SQ::Byte *)m_buffer + offset;
  76. sendUDPPackToTerminal(phoneNum, pRawData, bodyLen);
  77. dealDataSize += (sizeof(TCPHead)+bodyLen);
  78. }
  79. }
  80. //处理数据之后刷新缓冲区
  81. if (dealDataSize > 0)
  82. {
  83. memmove(m_buffer,m_buffer+dealDataSize,m_bytesReceived-dealDataSize);
  84. m_bytesReceived -= dealDataSize;
  85. }
  86. socket.async_read_some(boost::asio::buffer(m_buffer + m_bytesReceived, PACK_MAX_SIZE-m_bytesReceived),
  87. boost::bind(&TCPConnection::handleRead, shared_from_this(),
  88. boost::asio::placeholders::error,
  89. boost::asio::placeholders::bytes_transferred));
  90. return;
  91. }
  92. void TCPConnection::handleWrite( const boost::system::error_code& error, size_t bytesTransferred)
  93. {
  94. if( error != 0 )
  95. {
  96. std::cout << "write error." << std::endl;
  97. }
  98. }
  99. TCPServer::TCPServer(IoService& ioService, int port) :
  100. acceptor(ioService, tcp::endpoint(tcp::v4(), port)) {
  101. startAccept();
  102. }
  103. void TCPServer::startAccept() {
  104. boost::shared_ptr<TCPConnection> newConnection = TCPConnection::create(
  105. acceptor.get_io_service());
  106. acceptor.async_accept(newConnection->getSocket(),
  107. boost::bind(&TCPServer::handleAccept, this, newConnection,
  108. boost::asio::placeholders::error));
  109. }
  110. void TCPServer::handleAccept(boost::shared_ptr<TCPConnection> newConnection,
  111. const boost::system::error_code& error) {
  112. if (!error) {
  113. newConnection->start();
  114. startAccept();
  115. }
  116. }

客户端:

[cpp] view plain copy  
  1. class CommonSession : public boost::enable_shared_from_this<CommonSession>
  2. {
  3. //外部接口
  4. public:
  5. CommonSession( boost::asio::io_service *io_service, const std::string &ip, short port);
  6. void start();
  7. void sendDataBytes(SQ::Byte *pByte, int len);
  8. bool getConnectState() {return m_isConnected;}
  9. //内部接口
  10. protected:
  11. void onConnect( const boost::system::error_code &error );
  12. void onWrite( const boost::system::error_code& error, size_t bytes_transferred );
  13. void onRead( const boost::system::error_code& error, size_t bytes_transferred );
  14. private:
  15. tcp::socket                              m_socket;          //套接字
  16. char                                     m_buffer[PACK_MAX_SIZE];   //接收缓冲区
  17. bool                     m_isConnected;         //连接状态
  18. std::string                              m_ip;              //目标 I P
  19. short                                    m_port;            //目标端口
  20. size_t                                   m_bytesReceived;       //收字节数
  21. unsigned short                           m_serialNum;           //发流水号
  22. boost::recursive_mutex                   m_mutex;           //互斥体
  23. boost::asio::io_service*                 m_io_service;          //I O 服务
  24. };
[cpp] view plain copy  
  1. CommonSession::CommonSession( boost::asio::io_service *io_service, const std::string &ip, short port )
  2. : m_io_service( io_service ), m_socket( *io_service ), m_ip( ip ), m_port( port )
  3. {
  4. m_bytesReceived = 0;
  5. m_isConnected   = false;
  6. haveLog         = false;
  7. }
  8. void CommonSession::onConnect( const boost::system::error_code & error )
  9. {
  10. if( error )
  11. {
  12. //错误LOG只输出一次
  13. if (!haveLog)
  14. {
  15. haveLog = true;
  16. //连接失败
  17. std::cout << "connect to " << m_ip << ":" << m_port << " failed!" << " -> ERR : ";
  18. if( error.value() != boost::system::errc::operation_canceled )
  19. {
  20. std::cerr << boost::system::system_error( error ).what() << std::endl;
  21. }
  22. }
  23. //3秒后重新连接
  24. m_socket.close();
  25. boost::this_thread::sleep( boost::posix_time::seconds( 3 ) );
  26. start();
  27. }
  28. else
  29. {
  30. std::cout << "connect to " << m_ip << ":" << m_port << " successed! (time:"<< time(NULL)<<")" << std::endl;
  31. m_isConnected = true;
  32. static tcp::no_delay option( true );
  33. m_socket.set_option( option );
  34. m_socket.set_option( boost::asio::socket_base::keep_alive( true ) );
  35. //重发未发送的消息
  36. //发起读事件
  37. m_socket.async_read_some( boost::asio::buffer( m_buffer ),
  38. boost::bind( &CommonSession::onRead,
  39. shared_from_this(),
  40. boost::asio::placeholders::error,
  41. boost::asio::placeholders::bytes_transferred ) );
  42. }
  43. }
  44. void CommonSession::onWrite( const boost::system::error_code& error, size_t bytes_transferred )
  45. {
  46. if( error != 0 )
  47. {
  48. std::cout << "!!! send error, code is " << error<<" !!!"<<std::endl;
  49. std::cerr << boost::system::system_error( error ).what() << std::endl;
  50. }
  51. }
  52. void CommonSession::onRead( const boost::system::error_code& error, size_t bytes_transferred )
  53. {
  54. //cout <<"************************socket read " << bytes_transferred << " bytes codes ***************************"<< endl;
  55. //出错检查
  56. if( error != 0 )
  57. {
  58. std::cout << "!!! onRead error, code is " << error<<" !!!"<<std::endl;
  59. std::cerr << boost::system::system_error( error ).what() << std::endl;
  60. m_socket.close();
  61. //3秒后重新连接
  62. boost::this_thread::sleep( boost::posix_time::seconds( 3 ) );
  63. m_isConnected = false;
  64. start();
  65. return;
  66. }
  67. else if( bytes_transferred == 0 )
  68. {
  69. m_socket.close();
  70. //3秒后重新连接
  71. boost::this_thread::sleep( boost::posix_time::seconds( 3 ) );
  72. m_isConnected = false;
  73. start();
  74. return;
  75. }
  76. //解析消息（主要考虑到TCP分包和TCP粘包）
  77. size_t m_bytesReceived = 0;
  78. m_bytesReceived += bytes_transferred;
  79. size_t dealDataSize = 0;
  80. if (m_bytesReceived>=sizeof(TCPHead))
  81. {
  82. int offset = 0;
  83. while (true)//考虑到TCP粘包问题，做个循环
  84. {
  85. //剩余缓冲区不足消息头
  86. offset = dealDataSize;
  87. if (m_bytesReceived-offset < sizeof(SQ::TCPHead)) break;
  88. SQ::TCPHead* head = (SQ::TCPHead*)(m_buffer + offset);
  89. unsigned long phoneNum = SQ::Utils::BCD2long( head->terminalID, PHONENUM_LENGTH );
  90. offset += PHONENUM_LENGTH;
  91. unsigned short serialNum = ntohs( head->serialNum );
  92. offset += 2;
  93. unsigned char cmd = head->cmd ;
  94. offset += 1;
  95. int bodyLen = ntohl( head->bodyLen );
  96. offset += 4;
  97. if (bodyLen == 0)//处理来自网关的数据，而不是终端
  98. {
  99. std::cout<< "receive(time:"<< time(NULL)<<") command from gate, cmd:" << (int)cmd << " phoneNum:"<<phoneNum<<std::endl;
  100. dealCmd(cmd, phoneNum);
  101. dealDataSize += sizeof(SQ::TCPHead);
  102. }
  103. else//处理来自终端的消息
  104. {
  105. //剩余缓冲区不足消息体
  106. if (m_bytesReceived-offset < bodyLen) break;
  107. //取出808消息体
  108. char *pRawData = m_buffer + offset;
  109. //解开消息头
  110. CodecMsgHead csMsgHead;
  111. AbstractCodec tmpCode;
  112. tmpCode.decodeHead((SQ::Byte*)pRawData, &csMsgHead);
  113. //处理消息长度
  114. dealDataSize += (sizeof(SQ::TCPHead)+bodyLen);
  115. }
  116. }
  117. }
  118. //处理数据之后刷新缓冲区
  119. if (dealDataSize > 0)
  120. {
  121. memmove(m_buffer,m_buffer+dealDataSize,m_bytesReceived-dealDataSize);
  122. m_bytesReceived -= dealDataSize;
  123. }
  124. //发起读事件
  125. m_socket.async_read_some( boost::asio::buffer(m_buffer + m_bytesReceived, PACK_MAX_SIZE-m_bytesReceived),
  126. boost::bind( &CommonSession::onRead,
  127. shared_from_this(),
  128. boost::asio::placeholders::error,
  129. boost::asio::placeholders::bytes_transferred ) );
  130. }
  131. void CommonSession::start()
  132. {
  133. tcp::endpoint ep = tcp::endpoint( address::from_string( m_ip ), m_port );
  134. m_socket.async_connect( ep,
  135. boost::bind( &CommonSession::onConnect,
  136. shared_from_this(),
  137. boost::asio::placeholders::error ) );
  138. }
  139. //发送字节流，编码808协议之后的字节流
  140. void CommonSession::sendDataBytes(SQ::Byte *pByte, int len)
  141. {
  142. //检测连接
  143. if (!m_isConnected)
  144. {
  145. std::cout<< "connection(to "<<m_ip<<":"<< m_port<<") is not established, so sendDataBytes faild" <<std::endl;
  146. return;
  147. }
  148. //发送消息
  149. time_t nowTime = time(NULL);
  150. std::cout << "send:(time:"<<nowTime<<")" << SQ::Utils::bytes2HexString( pByte, len<=40?len:40);
  151. if (len<=40)
  152. {
  153. std::cout<< std::endl;
  154. }
  155. else
  156. {
  157. cout << "..." <<endl;
  158. }
  159. //  std::cout << "send:(time:"<<nowTime<<")" << SQ::Utils::bytes2HexString( pByte, len) <<endl;
  160. boost::asio::async_write(
  161. m_socket,
  162. boost::asio::buffer( pByte, len),
  163. boost::bind(&CommonSession::onWrite,shared_from_this(),boost::asio::placeholders::error,boost::asio::placeholders::bytes_transferred)
  164. );
  165. }

【Boost】boost库asio详解9——TCP的简单例子2相关推荐

  1. 【Boost】boost库asio详解8——TCP的简单例子1

    摘于boost官网的几个例子, 做了点小修改, 笔记之. 同步客户端 [cpp] view plain copy  print? void test_asio_synclient() { typede ...

  2. 【Boost】boost库asio详解9——UDP的简单例子1

    服务器: #include "stdafx.h" #include <iostream> #include <boost/asio.hpp> #includ ...

  3. 【Boost】boost库asio详解7——boost::asio::buffer用法

    1. asio::buffer常用的构造方法 asio::buffer有多种的构造方法,而且buffer大小是自动管理的 1.1 字符数组 [cpp] view plain copy  print? ...

  4. 【Boost】boost库asio详解3——io_service作为work pool

    无论如何使用,都能感觉到使用boost.asio实现服务器,不仅是一件非常轻松的事,而且代码很漂亮,逻辑也相当清晰,这点上很不同于ACE. 使用io_service作为处理工作的work pool,可 ...

  5. 【Boost】boost库asio详解5——resolver与endpoint使用说明

    tcp::resolver一般和tcp::resolver::query结合用,通过query这个词顾名思义就知道它是用来查询socket的相应信息,一般而言我们关心socket的东东有address ...

  6. 【Boost】boost库asio详解2——io_service::run函数无任务时退出的问题

    io_service::work类可以使io_service::run函数在没有任务的时候仍然不返回,直至work对象被销毁. [cpp] view plaincopy print? void tes ...

  7. 【Boost】boost库asio详解6——boost::asio::error的用法浅析

    1. 概述 一般而言我们创建用于接收error的类型大多声明如下: [cpp] view plain copy  print? boost::system::error_code error 我们用这 ...

  8. 【Boost】boost库asio详解4——deadline_timer使用说明

    deadline_timer和socket一样,都用io_service作为构造函数的参数.也即,在其上进行异步操作,都将导致和io_service所包含的iocp相关联.这同样意味着在析构 io_s ...

  9. 【Boost】boost库asio详解2——strand与io_service区别

    [cpp] view plain copy  print? namespace { // strand提供串行执行, 能够保证线程安全, 同时被post或dispatch的方法, 不会被并发的执行. ...

最新文章

  1. python入门视频教程推荐-python入门学习哪个书比较好(python视频教程知乎)
  2. shell脚本[] [[]] -n -z 的含义解析
  3. MVC设计模式学习总结
  4. poj 2388 排序的水题
  5. Spark源码分析之SparkContext
  6. Linux内核深入理解系统调用（3）：open 系统调用实现以及资源限制(setrlimit/getrlimit/prlimit)
  7. csrf 攻击及防御
  8. 奥威尔：老大哥在看着你-软件公司十诫
  9. vue中的阿里巴巴矢量图标使用
  10. 步进电机 高速光耦_光耦的参数以及高速光耦如何选型！-先进光半导体
  11. 时分秒倒计时的js实现
  12. 【小白向】利用笔记本+网线让台式机上网
  13. iOS 玩不转的GCD第0课时
  14. 【它山之石，可以攻玉】关于求职（实习）面试经验（2）
  15. Android studio 改app图标，名字及一键拨号
  16. 易快借的额度有多少？易快借的利息怎么计算？
  17. 认识图，用矩阵表示图
  18. 使用endnote来查找目标期刊
  19. Mac上NVM 安装与使用教程
  20. fastadmin input rule 自定义

热门文章

  1. RocketMQ集群之搭建2m2s集群（配置说明）
  2. BufferedReader_字符缓冲输入流
  3. 响应json数据之过滤静态资源
  4. Lambda标准格式
  5. 字符之间或者结构体之间比较
  6. 2018.7月Vue优质开源项目清单
  7. 极客班C++ STL（容器）第二周笔记
  8. JDK自带监控工具 jps、jinfo、jstat、jmap、jconsole
  9. [转]深入理解G1垃圾收集器
  10. 我的第一个IT产品：PublicLecture@HK【My First IT Product】