boost asio 异步实现tcp通讯

---恢复内容开始---

asioboost

目录(?)[-]

  1. 一前言
  2. 二实现思路
    1. 通讯包数据结构
    2. 连接对象
    3. 连接管理器
    4. 服务器端的实现
    5. 对象串行化

一、前言

boost asio可算是一个简单易用，功能又强大可跨平台的C++通讯库，效率也表现的不错，linux环境是epoll实现的，而windows环境是iocp实现的。而tcp通讯是项目当中经常用到通讯方式之一，实现的方法有各式各样，因此总结一套适用于自己项目的方法是很有必要，很可能下一个项目直接套上去就可以用了。

二、实现思路

1.通讯包数据结构

Tag：检查数据包是否合法，具体会在下面讲解；

Length：描述Body的长度；

Command：表示数据包的类型，0表示心跳包（长连接需要心跳来检测连接是否正常），1表示注册包（客户端连接上服务器之后要将相关信息注册给服务器），2表示业务消息包；

business_type：业务消息包类型，服务器会根据业务消息包类型将数据路由到对应的客户端（客户端是有业务类型分类的）；

app_id：客户端唯一标识符；

Data：消息数据；

2.连接对象

客户端连接上服务器之后，双方都会产生一个socket连接对象，通过这个对象可以收发数据，因此我定义为socket_session。

//socket_session.h

[cpp] view plaincopyprint?

  1. #pragma once
  2. #include <iostream>
  3. #include <list>
  4. #include <hash_map>
  5. #include <boost/bind.hpp>
  6. #include <boost/asio.hpp>
  7. #include <boost/shared_ptr.hpp>
  8. #include <boost/make_shared.hpp>
  9. #include <boost/thread.hpp>
  10. #include <boost/thread/mutex.hpp>
  11. #include <boost/enable_shared_from_this.hpp>
  12. #include <firebird/log/logger_log4.hpp>
  13. #include <firebird/detail/config.hpp>
  14. #include <firebird/socket_utils/message_archive.hpp>
  15. using boost::asio::ip::tcp;
  16. namespace firebird{
  17. enum command{ heartbeat = 0, regist, normal};
  18. const std::string tag = "KDS";
  19. class FIREBIRD_DECL socket_session;
  20. typedef boost::shared_ptr<socket_session> socket_session_ptr;
  21. class FIREBIRD_DECL socket_session:
  22. public boost::enable_shared_from_this<socket_session>,
  23. private boost::noncopyable
  24. {
  25. public:
  26. typedef boost::function<void(socket_session_ptr)> close_callback;
  27. typedef boost::function<void(
  28. const boost::system::error_code&,
  29. socket_session_ptr, message&)> read_data_callback;
  30. socket_session(boost::asio::io_service& io_service);
  31. ~socket_session(void);
  32. DWORD id() { return m_id; }
  33. WORD get_business_type(){ return m_business_type; }
  34. void set_business_type(WORD type) { m_business_type = type; }
  35. DWORD get_app_id(){ return m_app_id; }
  36. void set_app_id(DWORD app_id) { m_app_id = app_id; }
  37. std::string& get_remote_addr() { return m_name; }
  38. void set_remote_addr(std::string& name) { m_name = name; }
  39. tcp::socket& socket() { return m_socket; }
  40. void installCloseCallBack(close_callback cb){ close_cb = cb; }
  41. void installReadDataCallBack(read_data_callback cb) { read_data_cb = cb; }
  42. void start();
  43. void close();
  44. void async_write(const std::string& sMsg);
  45. void async_write(message& msg);
  46. bool is_timeout();
  47. void set_op_time(){std::time(&m_last_op_time);}
  48. private:
  49. static boost::detail::atomic_count m_last_id;
  50. DWORD m_id;
  51. WORD  m_business_type;
  52. DWORD m_app_id;
  53. std::string m_name;
  54. boost::array<char, 7> sHeader;
  55. std::string sBody;
  56. tcp::socket m_socket;
  57. boost::asio::io_service& m_io_service;
  58. std::time_t m_last_op_time;
  59. close_callback close_cb;
  60. read_data_callback read_data_cb;
  61. //发送消息
  62. void handle_write(const boost::system::error_code& e,
  63. std::size_t bytes_transferred, std::string* pmsg);
  64. //读消息头
  65. void handle_read_header(const boost::system::error_code& error);
  66. //读消息体
  67. void handle_read_body(const boost::system::error_code& error);
  68. void handle_close();
  69. };
  70. }

这里注意的是，定义了一个tag="KDS"，目的是为了检查收到的数据包是否有效，每一个数据包前3个字节不为“KDS”，那么就认为是非法的请求包，你也可以定义tag等于其它字符串，只要按协议发包就正常，当然这是比较简单的数据包检查方法了。比较严谨的方法是双方使用哈希算法来检查的，怎么做，这里先不做详解。

//socket_session.cpp

[cpp] view plaincopyprint?

  1. #include "socket_session.h"
  2. namespace firebird{
  3. boost::detail::atomic_count socket_session::m_last_id(0);
  4. socket_session::socket_session(boost::asio::io_service& io_srv)
  5. :m_io_service(io_srv), m_socket(io_srv),
  6. m_business_type(0), m_app_id(0)
  7. {
  8. m_id = ++socket_session::m_last_id;
  9. }
  10. socket_session::~socket_session(void)
  11. {
  12. m_socket.close();
  13. }
  14. void socket_session::start()
  15. {
  16. m_socket.set_option(boost::asio::ip::tcp::acceptor::linger(true, 0));
  17. m_socket.set_option(boost::asio::socket_base::keep_alive(true));
  18. std::time(&m_last_op_time);
  19. const boost::system::error_code error;
  20. handle_read_header(error);
  21. }
  22. void socket_session::handle_close()
  23. {
  24. try{
  25. m_socket.close();
  26. close_cb(shared_from_this());
  27. }
  28. catch(std::exception& e)
  29. {
  30. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "连接远程地址:[" << get_remote_addr() << "],socket异常:[" << e.what() << "]");
  31. }
  32. catch(...)
  33. {
  34. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "连接远程地址:[" << get_remote_addr() << "],socket异常:[未知异常]");
  35. }
  36. }
  37. void socket_session::close()
  38. {
  39. //由于回调中有加锁的情况，必须提交到另外一个线程去做，不然会出现死锁
  40. m_io_service.post(boost::bind(&socket_session::handle_close, shared_from_this()));
  41. }
  42. static int connection_timeout = 60;
  43. bool socket_session::is_timeout()
  44. {
  45. std::time_t now;
  46. std::time(&now);
  47. return now - m_last_op_time > connection_timeout;
  48. }
  49. //读消息头
  50. void socket_session::handle_read_header(const boost::system::error_code& error)
  51. {
  52. LOG4CXX_DEBUG(firebird_log, KDS_CODE_INFO  << "enter.");
  53. try{
  54. if(error)
  55. {
  56. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO  << "连接远程地址:[" << get_remote_addr() << "],socket异常:[" << error.message().c_str() << "]");
  57. close();
  58. return;
  59. }
  60. std::string data;
  61. data.swap(sBody);
  62. boost::asio::async_read(m_socket,
  63. boost::asio::buffer(sHeader),
  64. boost::bind(&socket_session::handle_read_body, shared_from_this(),
  65. boost::asio::placeholders::error));
  66. if (data.length() > 0 && data != "")
  67. {//读到数据回调注册的READ_DATA函数
  68. message msg;
  69. message_iarchive(msg, data);
  70. read_data_cb(error, shared_from_this(), msg);
  71. }
  72. }
  73. catch(std::exception& e)
  74. {
  75. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "连接远程地址:[" << get_remote_addr() << "],socket异常:[" << e.what() << "]");
  76. close();
  77. }
  78. catch(...)
  79. {
  80. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "连接远程地址:[" << get_remote_addr() << "],socket异常:[未知异常]");
  81. close();
  82. }
  83. }
  84. //读消息体
  85. void socket_session::handle_read_body(const boost::system::error_code& error)
  86. {
  87. LOG4CXX_DEBUG(firebird_log, KDS_CODE_INFO << "enter.");
  88. try{
  89. if(error)
  90. {
  91. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "连接远程地址:[" << get_remote_addr() << "],socket异常:[" << error.message().c_str() << "]");
  92. close();
  93. return;
  94. }
  95. if (tag.compare(0, tag.length(), sHeader.data(), 0, tag.length()))
  96. {
  97. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO <<  "连接远程地址:[" << get_remote_addr() << "],socket异常:[这是个非法连接!]");
  98. close();
  99. return;
  100. }
  101. DWORD dwLength = 0;
  102. char* len = (char*)&dwLength;
  103. memcpy(len, &sHeader[tag.length()], sizeof(dwLength));
  104. sBody.resize(dwLength);
  105. char* pBody = &sBody[0];
  106. boost::asio::async_read(m_socket,
  107. boost::asio::buffer(pBody, dwLength),
  108. boost::bind(&socket_session::handle_read_header, shared_from_this(),
  109. boost::asio::placeholders::error));
  110. }
  111. catch(std::exception& e)
  112. {
  113. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "连接远程地址:[" << get_remote_addr() << "],socket异常:[" << e.what() << "]");
  114. close();
  115. }
  116. catch(...)
  117. {
  118. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "连接远程地址:[" << get_remote_addr() << "],socket异常:[未知异常]");
  119. close();
  120. }
  121. }
  122. void socket_session::handle_write(const boost::system::error_code& error,
  123. std::size_t bytes_transferred, std::string* pmsg)
  124. {
  125. //数据发送成功就销毁
  126. if (pmsg != NULL)
  127. {
  128. delete pmsg;
  129. }
  130. if(error)
  131. {
  132. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "连接远程地址:[" << get_remote_addr() << "],socket异常:[" << error.message().c_str() << "]");
  133. close();
  134. return;
  135. }
  136. }
  137. void socket_session::async_write(const std::string& sMsg)
  138. {
  139. LOG4CXX_DEBUG(firebird_log, KDS_CODE_INFO  << "enter.")
  140. try
  141. {
  142. DWORD dwLength = sMsg.size();
  143. char* pLen = (char*)&dwLength;
  144. //由于是异步发送，要保证数据发送完整时，才把数据销毁
  145. std::string* msg = new std::string();
  146. msg->append(tag);
  147. msg->append(pLen, sizeof(dwLength));
  148. msg->append(sMsg);
  149. boost::asio::async_write(m_socket,boost::asio::buffer(*msg, msg->size()),
  150. boost::bind(&socket_session::handle_write, shared_from_this(),
  151. boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred,
  152. msg));
  153. }
  154. catch(std::exception& e)
  155. {
  156. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "连接远程地址:[" << get_remote_addr() << "],socket异常:[" << e.what() << "]");
  157. close();
  158. }
  159. catch(...)
  160. {
  161. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "连接远程地址:[" << get_remote_addr() << "],socket异常:[未知异常]");
  162. close();
  163. }
  164. }
  165. void socket_session::async_write(message& msg)
  166. {
  167. std::string data;
  168. message_oarchive(data, msg);
  169. async_write(data);
  170. }
  171. }

接受数据时，socket_session会先读取7个字节的head，比较前3个字节“KDS”，然后取得4个字节的Length，再读出Length长度的数据，最后将该数据传给read_data_cb回调函数处理，read_data_cb回调函数是在外部注册的。

3.连接管理器

对于服务器来说，它同时服务多个客户端，为了有效的管理，因此需要一个连接管理器，我定义为session_manager。session_manager主要是对socket_session的增删改查，和有效性检查。

//session_manager.h

[cpp] view plaincopyprint?

  1. #pragma once
  2. #include "socket_session.h"
  3. #include "filter_container.h"
  4. #include <boost/date_time/posix_time/posix_time.hpp>
  5. #include <boost/multi_index_container.hpp>
  6. #include <boost/multi_index/member.hpp>
  7. #include <boost/multi_index/ordered_index.hpp>
  8. #include <boost/typeof/typeof.hpp>
  9. #include <boost/random.hpp>
  10. #include <boost/pool/detail/singleton.hpp>
  11. namespace firebird{
  12. template<typename T>
  13. class var_gen_wraper
  14. {
  15. public:
  16. var_gen_wraper(): gen(boost::mt19937((boost::int32_t)std::time(0)),
  17. boost::uniform_smallint<>(1, 100)) {}
  18. typename T::result_type operator() () { return gen(); }
  19. private:
  20. T gen;
  21. };
  22. struct  session_stu
  23. {
  24. DWORD   id;
  25. WORD    business_type;
  26. std::string address;
  27. DWORD   app_id;
  28. socket_session_ptr session;
  29. };
  30. struct sid{};
  31. struct sbusiness_type{};
  32. struct saddress{};
  33. struct sapp_id{};
  34. enum session_idx_member{ session_id = 0, session_business_type, session_address, app_id};
  35. #define CLIENT 0
  36. #define SERVER 1
  37. typedef boost::multi_index::multi_index_container<
  38. session_stu,
  39. boost::multi_index::indexed_by<
  40. boost::multi_index::ordered_unique<
  41. boost::multi_index::tag<sid>, BOOST_MULTI_INDEX_MEMBER(session_stu, DWORD, id)>,
  42. boost::multi_index::ordered_non_unique<
  43. boost::multi_index::tag<sbusiness_type>, BOOST_MULTI_INDEX_MEMBER(session_stu, WORD, business_type)>,
  44. boost::multi_index::ordered_non_unique<
  45. boost::multi_index::tag<saddress>, BOOST_MULTI_INDEX_MEMBER(session_stu, std::string, address)>,
  46. boost::multi_index::ordered_non_unique<
  47. boost::multi_index::tag<sapp_id>, BOOST_MULTI_INDEX_MEMBER(session_stu, DWORD, app_id)>
  48. >
  49. > session_set;
  50. #define MULTI_MEMBER_CON(Tag) boost::multi_index::index<session_set,Tag>::type&
  51. #define MULTI_MEMBER_ITR(Tag) boost::multi_index::index<session_set,Tag>::type::iterator
  52. struct is_business_type {
  53. is_business_type(WORD type)
  54. :m_type(type)
  55. {
  56. }
  57. bool operator()(const session_stu& s)
  58. {
  59. return (s.business_type == m_type);
  60. }
  61. WORD m_type;
  62. };
  63. class session_manager
  64. {
  65. public:
  66. typedef boost::shared_lock<boost::shared_mutex> readLock;
  67. typedef boost:: unique_lock<boost::shared_mutex> writeLock;
  68. session_manager(boost::asio::io_service& io_srv, int type, int expires_time);
  69. ~session_manager();
  70. void add_session(socket_session_ptr p);
  71. void update_session(socket_session_ptr p);
  72. template<typename Tag, typename Member>
  73. void del_session(Member m)
  74. {
  75. writeLock lock(m_mutex);
  76. if (m_sessions.empty())
  77. {
  78. return ;
  79. }
  80. MULTI_MEMBER_CON(Tag) idx = boost::multi_index::get<Tag>(m_sessions);
  81. //BOOST_AUTO(idx, boost::multi_index::get<Tag>(m_sessions));
  82. BOOST_AUTO(iter, idx.find(m));
  83. if (iter != idx.end())
  84. {
  85. idx.erase(iter);
  86. }
  87. }
  88. //获取容器中的第一个session
  89. template<typename Tag, typename Member>
  90. socket_session_ptr get_session(Member m)
  91. {
  92. readLock lock(m_mutex);
  93. if (m_sessions.empty())
  94. {
  95. return socket_session_ptr();
  96. }
  97. MULTI_MEMBER_CON(Tag) idx = boost::multi_index::get<Tag>(m_sessions);
  98. BOOST_AUTO(iter, idx.find(m));
  99. return iter != boost::end(idx) ? iter->session : socket_session_ptr();
  100. }
  101. //随机获取容器中的session
  102. template<typename Tag>
  103. socket_session_ptr get_session_by_business_type(WORD m)
  104. {
  105. typedef filter_container<is_business_type, MULTI_MEMBER_ITR(Tag)> FilterContainer;
  106. readLock lock(m_mutex);
  107. if (m_sessions.empty())
  108. {
  109. return socket_session_ptr();
  110. }
  111. MULTI_MEMBER_CON(Tag) idx = boost::multi_index::get<Tag>(m_sessions);
  112. //对容器的元素条件过滤
  113. is_business_type predicate(m);
  114. FilterContainer fc(predicate, idx.begin(), idx.end());
  115. FilterContainer::FilterIter iter = fc.begin();
  116. if (fc.begin() == fc.end())
  117. {
  118. return socket_session_ptr();
  119. }
  120. //typedef boost::variate_generator<boost::mt19937, boost::uniform_smallint<>> var_gen;
  121. //typedef boost::details::pool::singleton_default<var_gen_wraper<var_gen>> s_var_gen;
  122. ////根据随机数产生session
  123. //s_var_gen::object_type &gen = s_var_gen::instance();
  124. //int step = gen() % fc.szie();
  125. int step = m_next_session % fc.szie();
  126. ++m_next_session;
  127. for (int i = 0; i < step; ++i)
  128. {
  129. iter++;
  130. }
  131. return iter != fc.end() ? iter->session : socket_session_ptr();
  132. }
  133. //根据类型和地址取session
  134. template<typename Tag>
  135. socket_session_ptr get_session_by_type_ip(WORD m, std::string& ip)
  136. {
  137. typedef filter_container<is_business_type, MULTI_MEMBER_ITR(Tag)> FilterContainer;
  138. readLock lock(m_mutex);
  139. if (m_sessions.empty())
  140. {
  141. return socket_session_ptr();
  142. }
  143. MULTI_MEMBER_CON(Tag) idx = boost::multi_index::get<Tag>(m_sessions);
  144. //对容器的元素条件过滤
  145. is_business_type predicate(m);
  146. FilterContainer fc(predicate, idx.begin(), idx.end());
  147. FilterContainer::FilterIter iter = fc.begin();
  148. if (fc.begin() == fc.end())
  149. {
  150. return socket_session_ptr();
  151. }
  152. while (iter != fc.end())
  153. {
  154. if (iter->session->get_remote_addr().find(ip) != std::string::npos)
  155. {
  156. break;
  157. }
  158. iter++;
  159. }
  160. return iter != fc.end() ? iter->session : socket_session_ptr();
  161. }
  162. //根据类型和app_id取session
  163. template<typename Tag>
  164. socket_session_ptr get_session_by_type_appid(WORD m, DWORD app_id)
  165. {
  166. typedef filter_container<is_business_type, MULTI_MEMBER_ITR(Tag)> FilterContainer;
  167. readLock lock(m_mutex);
  168. if (m_sessions.empty())
  169. {
  170. return socket_session_ptr();
  171. }
  172. MULTI_MEMBER_CON(Tag) idx = boost::multi_index::get<Tag>(m_sessions);
  173. //对容器的元素条件过滤
  174. is_business_type predicate(m);
  175. FilterContainer fc(predicate, idx.begin(), idx.end());
  176. FilterContainer::FilterIter iter = fc.begin();
  177. if (fc.begin() == fc.end())
  178. {
  179. return socket_session_ptr();
  180. }
  181. while (iter != fc.end())
  182. {
  183. if (iter->session->get_app_id() == app_id)
  184. {
  185. break;
  186. }
  187. iter++;
  188. }
  189. return iter != fc.end() ? iter->session : socket_session_ptr();
  190. }
  191. private:
  192. int m_type;
  193. int m_expires_time;
  194. boost::asio::io_service& m_io_srv;
  195. boost::asio::deadline_timer m_check_tick;
  196. boost::shared_mutex m_mutex;
  197. unsigned short m_next_session;
  198. session_set m_sessions;
  199. void check_connection();
  200. };
  201. }

这里主要用到了boost的multi_index容器，这是一个非常有用方便的容器，可实现容器的多列索引，具体的使用方法，在这里不多做详解。

//session_manager.cpp

[cpp] view plaincopyprint?

  1. #include "session_manager.h"
  2. namespace firebird{
  3. session_manager::session_manager(boost::asio::io_service& io_srv, int type, int expires_time)
  4. :m_io_srv(io_srv), m_check_tick(io_srv), m_type(type), m_expires_time(expires_time),m_next_session(0)
  5. {
  6. check_connection();
  7. }
  8. session_manager::~session_manager()
  9. {
  10. }
  11. //检查服务器所有session的连接状态
  12. void session_manager::check_connection()
  13. {
  14. try{
  15. writeLock lock(m_mutex);
  16. session_set::iterator iter = m_sessions.begin();
  17. while (iter != m_sessions.end())
  18. {
  19. LOG4CXX_DEBUG(firebird_log, "循环");
  20. if (CLIENT == m_type)//客户端的方式
  21. {
  22. if (!iter->session->socket().is_open())//已断开，删除已断开的连接
  23. {
  24. LOG4CXX_INFO(firebird_log, "重新连接[" << iter->address << "]");
  25. iter->session->close(); //通过关闭触发客户端重连
  26. }
  27. else{//连接中，发送心跳
  28. message msg;
  29. msg.command = heartbeat;
  30. msg.business_type = iter->session->get_business_type();
  31. msg.app_id = iter->session->get_app_id();
  32. msg.data() = "H";
  33. iter->session->async_write(msg);
  34. iter->session->set_op_time();
  35. }
  36. }
  37. else if (SERVER == m_type)//服务器的方式
  38. {
  39. if (!iter->session->socket().is_open())//已断开，删除已断开的连接
  40. {
  41. LOG4CXX_INFO(firebird_log, KDS_CODE_INFO << "删除已关闭的session:[" << iter->session->get_remote_addr() << "]");
  42. iter = m_sessions.erase(iter);
  43. continue;
  44. }
  45. else{//连接中，设定每30秒检查一次
  46. if (iter->session->is_timeout()) //如果session已长时间没操作，则关闭
  47. {
  48. LOG4CXX_INFO(firebird_log, KDS_CODE_INFO << "删除已超时的session:[" << iter->session->get_remote_addr() << "]");
  49. iter->session->close();//通过关闭触发删除session
  50. }
  51. }
  52. iter->session->set_op_time();
  53. }
  54. else{
  55. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "unknown manager_type");
  56. }
  57. ++iter;
  58. }
  59. LOG4CXX_DEBUG(firebird_log, "定时检查");
  60. m_check_tick.expires_from_now(boost::posix_time::seconds(m_expires_time));
  61. m_check_tick.async_wait(boost::bind(&session_manager::check_connection, this));
  62. }
  63. catch(std::exception& e)
  64. {
  65. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "[" << e.what() << "]");
  66. }
  67. catch(...)
  68. {
  69. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "unknown exception.");
  70. }
  71. }
  72. void session_manager::add_session(socket_session_ptr p)
  73. {
  74. writeLock lock(m_mutex);
  75. session_stu stuSession;
  76. stuSession.id = p->id();
  77. stuSession.business_type = 0;
  78. stuSession.address = p->get_remote_addr();
  79. stuSession.app_id = p->get_app_id();
  80. stuSession.session = p;
  81. m_sessions.insert(stuSession);
  82. }
  83. void session_manager::update_session(socket_session_ptr p)
  84. {
  85. writeLock lock(m_mutex);
  86. if (m_sessions.empty())
  87. {
  88. return ;
  89. }
  90. MULTI_MEMBER_CON(sid) idx = boost::multi_index::get<sid>(m_sessions);
  91. BOOST_AUTO(iter, idx.find(p->id()));
  92. if (iter != idx.end())
  93. {
  94. const_cast<session_stu&>(*iter).business_type = p->get_business_type();
  95. const_cast<session_stu&>(*iter).app_id = p->get_app_id();
  96. }
  97. }
  98. }

这个时候，我就可以使用id、business_type、address、app_id当做key来索引socket_session了，单使用map容器是做不到的。

还有索引时，需要的一个条件过滤器

//filter_container.h

[cpp] view plaincopyprint?

  1. #pragma once
  2. #include <boost/iterator/filter_iterator.hpp>
  3. namespace firebird{
  4. template <class Predicate, class Iterator>
  5. class filter_container
  6. {
  7. public:
  8. typedef boost::filter_iterator<Predicate, Iterator> FilterIter;
  9. filter_container(Predicate p, Iterator begin, Iterator end)
  10. :m_begin(p, begin, end),
  11. m_end(p, end, end)
  12. {
  13. }
  14. ~filter_container() {}
  15. FilterIter begin() { return m_begin; }
  16. FilterIter end()   { return m_end; }
  17. int szie() {
  18. int i = 0;
  19. FilterIter fi = m_begin;
  20. while(fi != m_end)
  21. {
  22. ++i;
  23. ++fi;
  24. }
  25. return i;
  26. }
  27. private:
  28. FilterIter m_begin;
  29. FilterIter m_end;
  30. };
  31. }

4.服务器端的实现

服务器我定义为server_socket_utils，拥有一个session_manager，每当accept成功得到一个socket_session时，都会将其增加到session_manager去管理，注册相关回调函数。

read_data_callback   接收到数据的回调函数

收到数据之后，也就是数据包的body部分，反序列化出command、business_type、app_id和data（我使用到了thrift），如果command==normal正常的业务包，会调用handle_read_data传入data。

close_callback 关闭socket_session触发的回调函数

根据id将该连接从session_manager中删除掉

//server_socket_utils.h

[cpp] view plaincopyprint?

  1. #pragma once
  2. #include "socket_session.h"
  3. #include "session_manager.h"
  4. #include <boost/format.hpp>
  5. #include <firebird/message/message.hpp>
  6. namespace firebird{
  7. using boost::asio::ip::tcp;
  8. class FIREBIRD_DECL server_socket_utils
  9. {
  10. private:
  11. boost::asio::io_service m_io_srv;
  12. boost::asio::io_service::work m_work;
  13. tcp::acceptor m_acceptor;
  14. void handle_accept(socket_session_ptr session, const boost::system::error_code& error);
  15. void close_callback(socket_session_ptr session);
  16. void read_data_callback(const boost::system::error_code& e,
  17. socket_session_ptr session, message& msg);
  18. protected:
  19. virtual void handle_read_data(message& msg, socket_session_ptr pSession) = 0;
  20. public:
  21. server_socket_utils(int port);
  22. ~server_socket_utils(void);
  23. void start();
  24. boost::asio::io_service& get_io_service() { return m_io_srv; }
  25. session_manager m_manager;
  26. };
  27. }

//server_socket_utils.cpp

[cpp] view plaincopyprint?

  1. #include "server_socket_utils.h"
  2. namespace firebird{
  3. server_socket_utils::server_socket_utils(int port)
  4. :m_work(m_io_srv),
  5. m_acceptor(m_io_srv, tcp::endpoint(tcp::v4(), port)),
  6. m_manager(m_io_srv, SERVER, 3)
  7. {
  8. //m_acceptor.set_option(boost::asio::ip::tcp::acceptor::reuse_address(true));
  9. //// 关闭连接前留0秒给客户接收数据
  10. //m_acceptor.set_option(boost::asio::ip::tcp::acceptor::linger(true, 0));
  11. //m_acceptor.set_option(boost::asio::ip::tcp::no_delay(true));
  12. //m_acceptor.set_option(boost::asio::socket_base::keep_alive(true));
  13. //m_acceptor.set_option(boost::asio::socket_base::receive_buffer_size(16384));
  14. }
  15. server_socket_utils::~server_socket_utils(void)
  16. {
  17. }
  18. void server_socket_utils::start()
  19. {
  20. try{
  21. socket_session_ptr new_session(new socket_session(m_io_srv));
  22. m_acceptor.async_accept(new_session->socket(),
  23. boost::bind(&server_socket_utils::handle_accept, this, new_session,
  24. boost::asio::placeholders::error));
  25. }
  26. catch(std::exception& e)
  27. {
  28. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "socket异常:[" << e.what() << "]");
  29. }
  30. catch(...)
  31. {
  32. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "socket异常:[未知异常]");
  33. }
  34. }
  35. void server_socket_utils::handle_accept(socket_session_ptr session, const boost::system::error_code& error)
  36. {
  37. if (!error)
  38. {
  39. try{
  40. socket_session_ptr new_session(new socket_session(m_io_srv));
  41. m_acceptor.async_accept(new_session->socket(),
  42. boost::bind(&server_socket_utils::handle_accept, this, new_session,
  43. boost::asio::placeholders::error));
  44. if (session != NULL)
  45. {
  46. //注册关闭回调函数
  47. session->installCloseCallBack(boost::bind(&server_socket_utils::close_callback, this, _1));
  48. //注册读到数据回调函数
  49. session->installReadDataCallBack(boost::bind(&server_socket_utils::read_data_callback, this, _1, _2, _3));
  50. boost::format fmt("%1%:%2%");
  51. fmt % session->socket().remote_endpoint().address().to_string();
  52. fmt % session->socket().remote_endpoint().port();
  53. session->set_remote_addr(fmt.str());
  54. session->start();
  55. m_manager.add_session(session);
  56. }
  57. }
  58. catch(std::exception& e)
  59. {
  60. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "socket异常:[" << e.what() << "]");
  61. }
  62. catch(...)
  63. {
  64. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "socket异常:[未知异常]");
  65. }
  66. }
  67. }
  68. void server_socket_utils::close_callback(socket_session_ptr session)
  69. {
  70. LOG4CXX_DEBUG(firebird_log, "close_callback");
  71. try{
  72. m_manager.del_session<sid>(session->id());
  73. }
  74. catch(std::exception& e)
  75. {
  76. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "socket异常:[" << e.what() << "]");
  77. }
  78. catch(...)
  79. {
  80. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "socket异常:[未知异常]");
  81. }
  82. }
  83. void server_socket_utils::read_data_callback(const boost::system::error_code& e,
  84. socket_session_ptr session, message& msg)
  85. {
  86. try{
  87. LOG4CXX_DEBUG(firebird_log, "command =[" << msg.command << "],["
  88. << msg.business_type << "],[" << msg.data() << "]");
  89. if (msg.command == heartbeat)
  90. {//心跳
  91. session->async_write(msg);
  92. }
  93. else if (msg.command == regist)
  94. {//注册
  95. session->set_business_type(msg.business_type);
  96. session->set_app_id(msg.app_id);
  97. m_manager.update_session(session);
  98. session->async_write(msg);
  99. LOG4CXX_FATAL(firebird_log, "远程地址:[" << session->get_remote_addr() << "],服务器类型:[" <<
  100. session->get_business_type() << "],服务器ID:[" << session->get_app_id() << "]注册成功!");
  101. }
  102. else if (msg.command == normal)
  103. {//业务数据
  104. handle_read_data(msg, session);
  105. }
  106. else
  107. {
  108. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "收到非法消息包!");
  109. }
  110. }
  111. catch(std::exception& e)
  112. {
  113. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "socket异常:[" << e.what() << "]");
  114. }
  115. catch(...)
  116. {
  117. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "socket异常:[未知异常]");
  118. }
  119. }
  120. }

5.客户端

客户端与服务器的逻辑也差不多，区别就是在于客户端通过connect得到socket_session，而服务器是通过accept得到socket_session。

//client_socket_utils.h

[cpp] view plaincopyprint?

  1. #pragma once
  2. #include "socket_session.h"
  3. #include "session_manager.h"
  4. #include <boost/algorithm/string.hpp>
  5. #include <firebird/message/message.hpp>
  6. namespace firebird{
  7. class FIREBIRD_DECL client_socket_utils
  8. {
  9. public:
  10. client_socket_utils();
  11. ~client_socket_utils();
  12. void session_connect(std::vector<socket_session_ptr>& vSession);
  13. void session_connect(socket_session_ptr pSession);
  14. //socket_session_ptr get_session(std::string& addr);
  15. boost::asio::io_service& get_io_service() { return m_io_srv; }
  16. protected:
  17. virtual void handle_read_data(message& msg, socket_session_ptr pSession) = 0;
  18. private:
  19. boost::asio::io_service m_io_srv;
  20. boost::asio::io_service::work m_work;
  21. session_manager m_manager;
  22. void handle_connect(const boost::system::error_code& error,
  23. tcp::resolver::iterator endpoint_iterator, socket_session_ptr pSession);
  24. void close_callback(socket_session_ptr session);
  25. void read_data_callback(const boost::system::error_code& e,
  26. socket_session_ptr session, message& msg);
  27. };
  28. }

//client_socket_utils.cpp

[cpp] view plaincopyprint?

  1. #include "client_socket_utils.h"
  2. namespace firebird{
  3. client_socket_utils::client_socket_utils()
  4. :m_work(m_io_srv), m_manager(m_io_srv, CLIENT, 3)
  5. {
  6. }
  7. client_socket_utils::~client_socket_utils()
  8. {
  9. }
  10. void client_socket_utils::session_connect(std::vector<socket_session_ptr>& vSession)
  11. {
  12. for (int i = 0; i < vSession.size(); ++i)
  13. {
  14. session_connect(vSession[i]);
  15. }
  16. }
  17. void client_socket_utils::session_connect(socket_session_ptr pSession)
  18. {
  19. std::string& addr = pSession->get_remote_addr();
  20. try{
  21. //注册关闭回调函数
  22. pSession->installCloseCallBack(boost::bind(&client_socket_utils::close_callback, this, _1));
  23. //注册读到数据回调函数
  24. pSession->installReadDataCallBack(boost::bind(&client_socket_utils::read_data_callback, this, _1, _2, _3));
  25. std::vector<std::string> ip_port;
  26. boost::split(ip_port, addr, boost::is_any_of(":"));
  27. if (ip_port.size() < 2)
  28. {
  29. //throw std::runtime_error("ip 格式不正确！");
  30. LOG4CXX_ERROR(firebird_log, "[" << addr << "] ip 格式不正确!");
  31. return;
  32. }
  33. tcp::resolver resolver(pSession->socket().get_io_service());
  34. tcp::resolver::query query(ip_port[0], ip_port[1]);
  35. tcp::resolver::iterator endpoint_iterator = resolver.resolve(query);
  36. //pSession->set_begin_endpoint(endpoint_iterator);//设置起始地址，以便重连
  37. //由于客户端是不断重连的，即使还未连接也要保存该session
  38. m_manager.add_session(pSession);
  39. tcp::endpoint endpoint = *endpoint_iterator;
  40. pSession->socket().async_connect(endpoint,
  41. boost::bind(&client_socket_utils::handle_connect, this,
  42. boost::asio::placeholders::error, ++endpoint_iterator, pSession));
  43. }
  44. catch(std::exception& e)
  45. {
  46. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "连接远程地址:[" << addr << "],socket异常:[" << e.what() << "]");
  47. }
  48. catch(...)
  49. {
  50. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "连接远程地址:[" << addr << "],socket异常:[未知异常]");
  51. }
  52. }
  53. void client_socket_utils::handle_connect(const boost::system::error_code& error,
  54. tcp::resolver::iterator endpoint_iterator, socket_session_ptr pSession)
  55. {
  56. LOG4CXX_DEBUG(firebird_log, KDS_CODE_INFO << " enter.");
  57. std::string sLog;
  58. try{
  59. if (!error)
  60. {
  61. LOG4CXX_FATAL(firebird_log, "服务器:[" << pSession->get_business_type() <<"],连接远程地址:[" << pSession->get_remote_addr().c_str() << "]成功!");
  62. pSession->start();
  63. //向服务器注册服务类型
  64. message msg;
  65. msg.command = regist;
  66. msg.business_type = pSession->get_business_type();
  67. msg.app_id = pSession->get_app_id();
  68. msg.data() = "R";
  69. pSession->async_write(msg);
  70. }
  71. else if (endpoint_iterator != tcp::resolver::iterator())
  72. {
  73. LOG4CXX_ERROR(firebird_log, "连接远程地址:[" << pSession->get_remote_addr().c_str() << "]失败，试图重连下一个地址。");
  74. pSession->socket().close();//此处用socket的close，不应用session的close触发连接，不然会导致一直重连
  75. tcp::endpoint endpoint = *endpoint_iterator;
  76. pSession->socket().async_connect(endpoint,
  77. boost::bind(&client_socket_utils::handle_connect, this,
  78. boost::asio::placeholders::error, ++endpoint_iterator, pSession));
  79. }
  80. else
  81. {
  82. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "连接远程地址:[" << pSession->get_remote_addr().c_str() << "]失败！");
  83. pSession->socket().close();//此处用socket的close，不应用session的close触发连接，不然会导致一直重连
  84. }
  85. }
  86. catch(std::exception& e)
  87. {
  88. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "连接远程地址:[" << pSession->get_remote_addr().c_str() <<"],socket异常:[" << e.what() << "]");
  89. }
  90. catch(...)
  91. {
  92. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "连接远程地址:[" << pSession->get_remote_addr().c_str() <<"],socket异常:[未知异常]");
  93. }
  94. }
  95. void client_socket_utils::read_data_callback(const boost::system::error_code& e,
  96. socket_session_ptr session, message& msg)
  97. {
  98. LOG4CXX_DEBUG(firebird_log, "command =[" << msg.command << "],["
  99. << msg.business_type << "],[" << msg.data() << "]");
  100. if (msg.command == heartbeat)
  101. {//心跳
  102. }
  103. else if (msg.command == regist)
  104. {//注册
  105. LOG4CXX_FATAL(firebird_log, "服务器:[" << session->get_business_type() <<"]注册成功。");
  106. }
  107. else if (msg.command == normal)
  108. {//业务数据
  109. handle_read_data(msg, session);
  110. }
  111. else
  112. {
  113. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "收到非法消息包!");
  114. }
  115. }
  116. //关闭session就会重连
  117. void client_socket_utils::close_callback(socket_session_ptr session)
  118. {
  119. LOG4CXX_DEBUG(firebird_log, KDS_CODE_INFO << "enter.");
  120. try{
  121. //tcp::resolver::iterator endpoint_iterator = context.session->get_begin_endpoint();
  122. std::string& addr = session->get_remote_addr();
  123. std::vector<std::string> ip_port;
  124. boost::split(ip_port, addr, boost::is_any_of(":"));
  125. if (ip_port.size() < 2)
  126. {
  127. LOG4CXX_ERROR(firebird_log, "[" << addr << "] ip 格式不正确!");
  128. return;
  129. }
  130. tcp::resolver resolver(session->socket().get_io_service());
  131. tcp::resolver::query query(ip_port[0], ip_port[1]);
  132. tcp::resolver::iterator endpoint_iterator = resolver.resolve(query);
  133. tcp::endpoint endpoint = *endpoint_iterator;
  134. session->socket().async_connect(endpoint,
  135. boost::bind(&client_socket_utils::handle_connect, this,
  136. boost::asio::placeholders::error, ++endpoint_iterator, session));
  137. }
  138. catch(std::exception& e)
  139. {
  140. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "连接远程地址:[" << session->get_remote_addr().c_str() <<"],socket异常:[" << e.what() << "]");
  141. }
  142. catch(...)
  143. {
  144. LOG4CXX_ERROR(firebird_log, KDS_CODE_INFO << "连接远程地址:[" << session->get_remote_addr().c_str() <<"],socket异常:[未知异常]");
  145. }
  146. }
  147. }

5.对象串行化

socket_session发送和接收数据包的时候使用到了对象串行化，我这里是通过thrift实现的，其实boost的serialization库也提供了这样的功能，使用起来更为方便，但我在测试过程中，thrift相比之下性能会高很多，因此就坚持使用thrift了，感兴趣的话可以看我之前写的《使用thrift串行化对象》和《轻量级序列化库boost serialization》。

5.1字符串与thrift对象的相互转换

[cpp] view plaincopyprint?

  1. #pragma once
  2. #include <boost/shared_ptr.hpp>
  3. #include <transport/TBufferTransports.h>
  4. #include <protocol/TProtocol.h>
  5. #include <protocol/TBinaryProtocol.h>
  6. namespace firebird{
  7. using namespace apache::thrift;
  8. using namespace apache::thrift::transport;
  9. using namespace apache::thrift::protocol;
  10. template<typename T>
  11. void thrift_iserialize(T& stu, std::string& s)
  12. {
  13. boost::shared_ptr<TMemoryBuffer> trans(new TMemoryBuffer((uint8_t*)&s[0], s.size()));
  14. boost::shared_ptr<TProtocol> proto(new TBinaryProtocol(trans));
  15. stu.read(proto.get());
  16. }
  17. template<typename T>
  18. void thrift_oserialize(T& stu, std::string& s)
  19. {
  20. boost::shared_ptr<TMemoryBuffer> trans(new TMemoryBuffer());
  21. boost::shared_ptr<TProtocol> proto(new TBinaryProtocol(trans));
  22. stu.write(proto.get());
  23. s = trans->getBufferAsString();
  24. }
  25. }

5.2通过thrift对象，普通的对象与字符串的相互转换

[cpp] view plaincopyprint?

  1. #pragma once
  2. #include "message_archive.hpp"
  3. #include <firebird/archive/thrift_archive.hpp>
  4. #include <firebird/message/TMessage_types.h>
  5. namespace firebird
  6. {
  7. /*** message to ThriftMessage ***/
  8. void msg_to_tmsg(TMessage& tmsg, message& msg)
  9. {
  10. //设置
  11. tmsg.command = msg.command;
  12. tmsg.business_type = msg.business_type;
  13. tmsg.app_id = msg.app_id;
  14. //设置context
  15. tmsg.context.cmdVersion = msg.context().cmdVersion;
  16. tmsg.context.cpid.swap(msg.context().cpid);
  17. tmsg.context.remote_ip.swap(msg.context().remote_ip);
  18. tmsg.context.wSerialNumber = msg.context().wSerialNumber;
  19. tmsg.context.session_id = msg.context().session_id;
  20. //设置source
  21. for (int i = 0; i < msg.source().size(); ++i)
  22. {
  23. tmsg.source.push_back(msg.source()[i]);
  24. }
  25. //设置destination
  26. for (int i = 0; i < msg.destination().size(); ++i)
  27. {
  28. tmsg.destination.push_back(msg.destination()[i]);
  29. }
  30. //设置data
  31. tmsg.data = msg.data();
  32. }
  33. /*** ThriftMessage to message ***/
  34. void tmsg_to_msg(message& msg, TMessage& tmsg)
  35. {
  36. //设置
  37. msg.command = tmsg.command;
  38. msg.business_type = tmsg.business_type;
  39. msg.app_id = tmsg.app_id;
  40. //设置context
  41. msg.context().cmdVersion = tmsg.context.cmdVersion;
  42. msg.context().cpid = tmsg.context.cpid;
  43. msg.context().remote_ip = tmsg.context.remote_ip;
  44. msg.context().wSerialNumber = tmsg.context.wSerialNumber;
  45. msg.context().session_id = tmsg.context.session_id;
  46. //设置source
  47. for (int i = 0; i < tmsg.source.size(); ++i)
  48. {
  49. msg.source() << tmsg.source[i];
  50. }
  51. //设置destination
  52. for (int i = 0; i < tmsg.destination.size(); ++i)
  53. {
  54. msg.destination() << tmsg.destination[i];
  55. }
  56. //设置data
  57. msg.data() = tmsg.data;
  58. }
  59. void message_iarchive(message& msg, std::string& s)
  60. {
  61. TMessage tmsg;
  62. thrift_iserialize(tmsg, s);
  63. tmsg_to_msg(msg, tmsg);
  64. }
  65. void message_oarchive(std::string& s, message& msg)
  66. {
  67. TMessage tmsg;
  68. msg_to_tmsg(tmsg, msg);
  69. thrift_oserialize(tmsg, s);
  70. }
  71. }

---恢复内容结束---

boost asio 异步实现tcp通讯

时间： 2024-11-10 13:50:35

boost asio 异步实现tcp通讯的相关文章

Boost Asio 异步TCP服务器框架

Boost Asio 异步TCP服务器框架 flyfish 2015-5-30 session 类 头文件 #pragma once #include <boost/asio.hpp> #include <boost/bind.hpp> #include <boost/shared_ptr.hpp> #include <boost/enable_shared_from_this.hpp> class session: public boost::enable

boost asio异步读写网络聊天程序客户端 实例详解

// // chat_client.cpp // ~~~~~~~~~~~~~~~ // // Copyright (c) 2003-2013 Christopher M. Kohlhoff (chris at kohlhoff dot com) // // Distributed under the Boost Software License, Version 1.0. (See accompanying // file LICENSE_1_0.txt or copy at http://ww

10 C++ Boost ASIO网路通信库 TCP/UDP,HTTP

  tcp 同步服务器,显示服务器端时间 tcp 同步服务器,提供多种选择 多线程的tcp 同步服务器 tcp 同步客户端 boost 域名地址解析 tcp异步服务器 tcp 异步客户端 UDP同步服务器 UDP同步客户端 UDP异步服务器 UDP异步客户端 HTTP同步客户端 HTTP异步客户端 同步实验: 异步实验 多线程异步实验 tcp 同步服务器,显示服务器端时间 [email protected]:~/boost$ cat main.cpp  #include <ctime> #in

stream opencv mat (webcam)frame throught tcp by boost asio

客户端: ? 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84

boost asio 学习(九) boost::asio 网络封装

http://www.gamedev.net/blog/950/entry-2249317-a-guide-to-getting- started-with-boostasio?pg=10 9. A boost::asio network wrapper (TCP) 现在我们了解asio和TCP网络方面的知识,我们可以尝试下封装网络底层.通过使用这个封装,我们可以重用代码并且将精力集中于业务逻 辑方面而不在网络通讯方面花费太多精力. 重要提示:本代码仅仅用于教学目的.不要在商业系统中使用该代码,

Coroutine协程库：网络性能完爆Asio异步模型

在purecpp社区的github组织中有一个协程库:https://github.com/topcpporg/cpp_features 近日有用户找到我,想要了解一下coroutine库在网络方面的性能,于是选取已入选标准库的boost.asio网络库的异步模型做了一次对比测试,结果完爆asio异步模型,coroutine库网络性能是asio异步模型的2-14倍,使用8线程处理小包时,QPS可达670万/秒. 简要介绍一下测试流程: 1.TCP协议,C/S是同一台物理机的两个进程,建立1000

Boost.Asio c++ 网络编程翻译（14）

保持活动 假如,你需要做下面的操作: io_service service; ip::tcp::socket sock(service); char buff[512]; ... read(sock, buffer(buff)); 在这个例子中,sock和buff的存在时间都必须比read()调用的时间要长.也就是说,在调用read()返回之前,它们都必须有效.这就是你期望的:你传给一个方法的所有参数在参数内部都必须有效.当我们采用异步方式时,事情会变得越复杂. io_service servi

boost.asio系列——socket编程

asio的主要用途还是用于socket编程,本文就以一个tcp的daytimer服务为例简单的演示一下如何实现同步和异步的tcp socket编程. 客户端 客户端的代码如下: #include <iostream>    #include <boost/array.hpp>    #include <boost/asio.hpp> using boost::asio::ip::tcp; int main(int argc, char* argv[])    {    

Boost.Asio基本原理（CSDN也有Markdown了，好开森）

Boost.Asio基本原理 这一章涵盖了在使用Boost.Asio时必须知道的一些事情.我们也将深入研究比同步编程更复杂.更有乐趣的异步编程. 网络API 这一部分包含了当使用Boost.Asio编写网络应用程序时必须知道的事情. Boost.Asio命名空间 Boost.Asio的所有内容都包含在boost::asio命名空间或者其子命名空间内. * boost::asio:这是核心类和函数所在的地方.重要的类有io_service和streambuf.类似read, read_at, re