boost asio tcp server 拆分

Posted 2020-07-19 独孤酷酷

从官方给出的示例中对于 boost::asio::ip::tcp::acceptor 类的使用，是直接使用构造函数进行构造对象，这一种方法用来学习是一个不错的方式。

但是要用它来做项目却是不能够满足我们的需求的，可它有相应的接口，可以让我们更灵活的使用它来做我们的项目。我们可以把这个accptor 的使用拆分开来，就是分成几个步骤来做。这样我们就可以在我们的项目中，在多个函数里面对它进行一步一步的生成。

 

简单的用法：

  1 #include <iostream>
  2 
  3 #include <boost/asio/io_service.hpp>
  4 #include <boost/shared_ptr.hpp>
  5 #include <boost/shared_array.hpp>
  6 #include <boost/make_shared.hpp>
  7 #include <boost/function.hpp>
  8 #include <boost/bind.hpp>
  9 #include <boost/asio/placeholders.hpp>
 10 #include <boost/asio/read.hpp>
 11 #include <boost/asio/write.hpp>
 12 #include <boost/asio/io_service.hpp>
 13 #include <boost/asio/ip/tcp.hpp>
 14 
 15 void async_accept();
 16 void handle_accept(boost::shared_ptr<boost::asio::ip::tcp::socket> new_conn, 
 17                    const boost::system::error_code &ec);
 18 void async_read(boost::shared_ptr<boost::asio::ip::tcp::socket> conn);
 19 void handle_msg(
 20     boost::shared_ptr<boost::asio::ip::tcp::socket> conn, 
 21     boost::shared_array<char> sa_len, 
 22     const boost::system::error_code &ec,
 23     std::size_t bytes_transfered);
 24 
 25 // 这里将这些对象，写为全局的，在实际的代码中，应该是一个类的成员变量或者其他方式存在
 26 boost::asio::io_service io_svc;    // io service 实例
 27 boost::asio::ip::address_v4 lis_ip;    // 默认监听本机所有IP
 28 boost::asio::ip::tcp::endpoint lis_ep(lis_ip, 20017);
 29 boost::asio::ip::tcp::acceptor acceptor(io_svc, lis_ep);
 30 
 31 int main(int argc, char *argv[])
 32 {
 33     async_accept();
 34 
 35     io_svc.run();
 36 
 37     return 0;
 38 }
 39 
 40 void async_accept()
 41 {
 42     boost::shared_ptr<boost::asio::ip::tcp::socket> new_sock
 43         = boost::make_shared<boost::asio::ip::tcp::socket>(boost::ref(io_svc));
 44     acceptor.async_accept(*new_sock, 
 45         boost::bind(handle_accept, new_sock, boost::asio::placeholders::error) );
 46 }
 47 
 48 void handle_accept(boost::shared_ptr<boost::asio::ip::tcp::socket> new_conn, 
 49                    const boost::system::error_code &ec)
 50 {
 51     if (ec != 0)
 52     {
 53         //LOG_INFO(get_logger(), "accept failed: " << ec.message());
 54         std::cout << "accept failed: " << ec.message() << std::endl;
 55         return ;
 56     }
 57     //LOG_INFO(get_logger(), "a new client connected." << new_conn->remote_endpoint());
 58     std::cout << "a new client connected." << new_conn->remote_endpoint() << std::endl;
 59 
 60     async_read(new_conn);
 61 
 62     // 处理下一个连接，每次处理完了之后，需要再次accept。
 63     // 否则BOOST 将只处理一次，然后结束监听。
 64     // 所以这里可以处理一个情况，就是当你要结束监听的时候，只要在这里return 
 65     // 那么io_service 的run() 函数就结束监听。但如果有其他的异步操作时，
 66     // run() 函数还是会继续运行的。
 67     async_accept();
 68 }
 69 
 70 void async_read(boost::shared_ptr<boost::asio::ip::tcp::socket> conn)
 71 {
 72     static const int PACKAGE_LENGTH = 6;
 73     // 数据报文长度为 6个字节
 74     boost::shared_array<char> sa_len(new char[PACKAGE_LENGTH]);
 75 
 76     // 回调函数
 77     boost::function<void (const boost::system::error_code &, std::size_t)> cb_msg_len;
 78     cb_msg_len = boost::bind(handle_msg, conn, sa_len, _1, _2);
 79 
 80     // 异步读，读一个报文的长度，boost::asio::async_read() 函数有个特点，
 81     // 它会将这里指定的buffer 缓冲区读满了才会去回调handle_msg 函数，
 82     boost::asio::async_read(*conn, 
 83         boost::asio::buffer(sa_len.get(), PACKAGE_LENGTH), cb_msg_len);
 84 
 85 }
 86 
 87 void handle_msg(
 88     boost::shared_ptr<boost::asio::ip::tcp::socket> conn, 
 89     boost::shared_array<char> sa_len, 
 90     const boost::system::error_code &ec,
 91     std::size_t bytes_transfered)
 92 {
 93     if (!conn->is_open())
 94     {
 95         //LOG_INFO(g_logger, "socket was not opened.");
 96         std::cout << "socket was not opened." << std::endl;
 97         //handle_dis_connect(conn);
 98         return ;
 99     }
100 
101     if (ec != 0)
102     {
103         if (ec == boost::asio::error::eof)
104             std::cout << "Disconnect from " << conn->remote_endpoint() << std::endl;
105         else
106             std::cout << "Error on receive: " << ec.message() << std::endl;
107 
108         //handle_dis_connect(the_conn);
109         return ;
110     }
111 
112     // 这里对接收到的数据做处理
113     // ...
114 
115     // 处理完了之后，类似accept 的异常调用一样，需要继续调用异步的读数据
116     // 同样的，如果要结束一个连接，正常的结束应该在这里return 调用。
117     // 当然了，使用socket 的close() shut_down() 函数也可以关闭这个连接。
118     async_read(conn);
119 }

View Code

 

发现个严重的问题，asio_server 类是个单独的类，而在这个类的成员函数中对this 指针做了操作，而且是将该指针以参数的形式传递给了一个函数指针。这样做是非常危险的，如果在main() 函数中实例化了一个asio_server 对象然后传入到io_service 中，然后asio_server 的实例生命周期到了，那么它的this 指针 则是一个野 指针，这样在asio_serivce 中却有可能 还在使用它，那我们就知道会发生什么危险的事情了。所以下面的代码不可取，我们可以将asio_server 派生自boost::enable_shared_from_this<asio_server> 如果需要还可以派生自boost::noncopyable，然后在用到this 的地方使用成boost::enable_shared_from_this 的一个成员函数 shared_from_this() 来获取到该对象的一个shred_ptr 对象，当然了对于asio_server 也需要用boost::shared_ptr 来构造。

灵活的用法：

 1 // main.cpp
 2 
 3 #include <iostream>
 4 
 5 #include <boost/asio/io_service.hpp>
 6 
 7 #include "asio_server.h"
 8 
 9 #include <middle/base/log.h>
10 
11 log4cplus::Logger        g_logger;
12 
13 int main(int argc, char *argv[])
14 {
15     if (argc < 2)
16     {
17         std::cout << "Usage: " << argv[0] << " <config>." << std::endl;
18         return -1;
19     }
20 
21     const char *str_log_cfg = argv[1];
22     const std::string APPENDER_NAME = "pack_back_svr";
23     log4cplus::PropertyConfigurator::doConfigure(LOG4CPLUS_TEXT(str_log_cfg));
24     g_logger = log4cplus::Logger::getInstance(LOG4CPLUS_TEXT(APPENDER_NAME));
25 
26 
27     boost::asio::io_service io_svc;    // io service 实例
28 
29     asio_server server_01(io_svc);
30     server_01.listen("", "32500");
31 
32 
33     io_svc.run();
34 
35     return 0;
36 }

View Code

 1 // asio_server.h
 2 #ifndef ASIO_SERVER_H
 3 #define ASIO_SERVER_H
 4 
 5 #include <boost/asio/io_service.hpp>
 6 #include <boost/asio/ip/tcp.hpp>
 7 #include <boost/shared_array.hpp>
 8 
 9 class asio_server
10 {
11 public:
12     asio_server(boost::asio::io_service &io_svc);
13     ~asio_server();
14 
15     bool listen(const char *lis_ip, const char *lis_port);
16     bool listen(const char *lis_ip, unsigned short lis_port);
17     bool listen(const char *lis_port);
18     bool listen(unsigned short lis_port);
19 
20 private:
21     void async_accept();
22     void handle_accept(boost::shared_ptr<boost::asio::ip::tcp::socket> sp_new_sock, 
23         const boost::system::error_code &ec);
24 
25     void handle_new_connect(boost::shared_ptr<boost::asio::ip::tcp::socket> sp_new_sock);
26     void start_read(boost::shared_ptr<boost::asio::ip::tcp::socket> sp_new_sock);
27 
28     void handle_msg(
29         boost::shared_ptr<boost::asio::ip::tcp::socket> the_conn, 
30         boost::shared_array<char> sa_len, 
31         const boost::system::error_code &ec,
32         std::size_t bytes_transfered);
33 
34 private:
35     boost::asio::io_service                &m_io_svc;
36     boost::asio::ip::tcp::acceptor         m_acceptor;
37 };
38 
39 
40 #endif

View Code

  1 // asio_server.cpp
  2 #include "asio_server.h"
  3 
  4 #include <stdio.h>
  5 #include <stdlib.h>
  6 #include <iostream>
  7 
  8 #ifdef WIN32
  9 
 10 #    include <windows.h>
 11 
 12 // windows 下没有snprintf 使用_snprintf 替换
 13 #    ifndef snprintf
 14 #        define snprintf _snprintf
 15 #    endif
 16 
 17 #endif
 18 
 19 #include <boost/shared_ptr.hpp>
 20 #include <boost/make_shared.hpp>
 21 #include <boost/function.hpp>
 22 #include <boost/bind.hpp>
 23 #include <boost/asio/placeholders.hpp>
 24 #include <boost/asio/read.hpp>
 25 #include <boost/asio/write.hpp>
 26 
 27 #include <middle/base/log.h>
 28 
 29 extern log4cplus::Logger        g_logger;
 30 
 31 asio_server::asio_server(boost::asio::io_service &io_svc)
 32     : m_io_svc(io_svc)
 33     , m_acceptor(m_io_svc)
 34 {
 35 
 36 }
 37 
 38 asio_server::~asio_server()
 39 {
 40 
 41 }
 42 
 43 bool asio_server::listen(const char *lis_ip, const char *lis_port)
 44 {
 45     if (lis_port == NULL || lis_port[0] == 0)
 46     {
 47         LOG_ERROR(g_logger, "listen port is NULL.");
 48         return false;
 49     }
 50 
 51     unsigned short us_port = 0;
 52     if (sscanf(lis_port, "%hu", &us_port) != 1)
 53     {
 54         // printf("Error: listen port <==> \\"%s\\"", lis_port);
 55         LOG_ERROR(g_logger, "listen port: " << lis_port);
 56         return false;
 57     }
 58 
 59     return listen(lis_ip, us_port);
 60 }
 61 bool asio_server::listen(const char *lis_ip, unsigned short lis_port)
 62 {
 63     boost::asio::ip::address_v4 lis_addr;
 64     if (lis_ip != NULL && lis_ip[0] != 0)
 65         lis_addr = boost::asio::ip::address_v4::from_string(lis_ip);
 66 
 67     boost::asio::ip::tcp::endpoint lis_ep;
 68     lis_ep.address(lis_addr);
 69     lis_ep.port(lis_port);
 70 
 71     boost::system::error_code ec_temp;
 72     m_acceptor.open(boost::asio::ip::tcp::v4());
 73     m_acceptor.bind(lis_ep, ec_temp);
 74     if (ec_temp != 0)
 75     {
 76         LOG_ERROR(g_logger, "bind failed: " << ec_temp.message());
 77         return false;
 78     }
 79     m_acceptor.listen();
 80 
 81     LOG_INFO(g_logger, "listen ==> " << m_acceptor.local_endpoint());
 82 
 83     async_accept();
 84 
 85     return true;
 86 }
 87 bool asio_server::listen(const char *lis_port)
 88 {
 89     unsigned short us_port = 0;
 90     sscanf(lis_port, "%hu", &us_port);
 91 
 92     return listen("", us_port);
 93 }
 94 bool asio_server::listen(unsigned short lis_port)
 95 {
 96     return listen("", lis_port);
 97 }
 98 
 99 void asio_server::async_accept()
100 {
101     try
102     {
103         boost::shared_ptr<boost::asio::ip::tcp::socket> new_sock
104             = boost::make_shared<boost::asio::ip::tcp::socket>(boost::ref(m_io_svc));
105 
106         boost::function<void (const boost::system::error_code& error)> cb_accept;
107         cb_accept = boost::bind(&asio_server::handle_accept, this, 
108             new_sock, boost::asio::placeholders::error);
109 
110         m_acceptor.async_accept(*new_sock, cb_accept);
111     }
112     catch (boost::system::system_error& e)
113     {
114         LOG_ERROR(g_logger, "Exception system_error: " << e.what());
115     }
116     catch (std::exception& e)
117     {
118         LOG_ERROR(g_logger, "Exception std: " << e.what());
119     }
120     catch (...)
121     {
122         LOG_ERROR(g_logger, "Unknown exception.");
123     }
124 }
125 
126 void asio_server::handle_accept(
127     boost::shared_ptr<boost::asio::ip::tcp::socket> sp_new_sock, 
128     const boost::system::error_code &ec)
129 {
130     if (ec != 0)
131     {
132         LOG_INFO(g_logger, "accept failed: " << ec.message());
133         return ;
134     }
135     LOG_INFO(g_logger, "a new client connected." << sp_new_sock->remote_endpoint());
136 
137     handle_new_connect(sp_new_sock);
138     start_read(sp_new_sock);
139 
140     // 处理下一个连接
141     async_accept();
142 }
143 
144 void asio_server::handle_new_connect(
145     boost::shared_ptr<boost::asio::ip::tcp::socket> sp_new_sock)
146 {
147     LOG_INFO(g_logger, "handle_new_connect function");
148 }
149 
150 void asio_server::start_read(
151     boost::shared_ptr<boost::asio::ip::tcp::socket> sp_new_sock)
152 {
153     static const int PACKAGE_LENGTH = 6;
154 
155     // 数据报文长度为 6个字节
156     boost::shared_array<char> sa_len(new char[PACKAGE_LENGTH]);
157 
158     // 回调函数
159     boost::function<void (const boost::system::error_code &, std::size_t)> cb_msg_len;
160     cb_msg_len = boost::bind(&asio_server::handle_msg, this, 
161         sp_new_sock, sa_len, _1, _2);
162 
163     // 异步读，读一个报文的长度
164     boost::asio::async_read(*sp_new_sock, 
165         boost::asio::buffer(sa_len.get(), PACKAGE_LENGTH), cb_msg_len);
166 }
167 
168 void asio_server::handle_msg(
169     boost::shared_ptr<boost::asio::ip::tcp::socket> the_conn, 
170     boost::shared_array<char> sa_len, 
171     const boost::system::error_code &ec,
172     std::size_t bytes_transfered)
173 {
174     LOG_INFO(g_logger, "handle message");
175 }

View Code

 这里面用到了log4cplus 日志库，主要是用来写日志的，将日志相关的代码用std::cout  替换就可以直接编译运行了。

当然了客户端还需要另外写。

不过你可以在网上找一些TCP 调试工具来测试一下。