1 // server.cpp 2 3 #if 0 4 多个线程对同一个io_service 对象处理 5 用到第三方库：log4cplus, google::protobuf 6 用到C++11的特性，Windows 需要用到vs2013 gcc 4.8 7 #endif 8 9 #include <iostream> 10 #include <thread> 11 #include <vector> 12 13 #include <boost/asio.hpp> 14 15 #include <boost/shared_array.hpp> 16 #include <boost/make_shared.hpp> 17 #include <boost/function.hpp> 18 #include <boost/bind.hpp> 19 20 #include <common.pb.h> 21 22 void async_accept(); 23 void handle_accept(boost::shared_ptr<boost::asio::ip::tcp::socket> new_conn, 24 const boost::system::error_code &ec); 25 void async_read_head(boost::shared_ptr<boost::asio::ip::tcp::socket> conn); 26 void handle_head( 27 boost::shared_ptr<boost::asio::ip::tcp::socket> conn, 28 boost::shared_array<char> sa_len, 29 const boost::system::error_code &ec, 30 std::size_t bytes_transfered); 31 void async_read_proto(boost::shared_ptr<boost::asio::ip::tcp::socket> conn, int32_t len); 32 void handle_proto( 33 boost::shared_ptr<boost::asio::ip::tcp::socket> conn, 34 boost::shared_array<char> sa_data, 35 const boost::system::error_code &ec, 36 std::size_t bytes_transfered); 37 38 boost::asio::io_service io_svc; 39 boost::asio::ip::address_v4 lis_ip; // 默认监听本机所有IP 40 boost::asio::ip::tcp::endpoint lis_ep(lis_ip, 20017); 41 boost::asio::ip::tcp::acceptor acceptor(io_svc, lis_ep); 42 43 #include <Log.h> // log4cplus 的相关头文件，这里不再一个一个敲出来了。 44 45 log4cplus::Logger *gLog = nullptr; 46 47 static const int PACKAGE_LENGTH = 16; 48 49 int main(int argc, char *argv[]) 50 { 51 log4cplus::initialize(); 52 53 static log4cplus::Logger s_log = log4cplus::Logger::getInstance("server"); 54 gLog = &s_log; 55 56 LOG4CPLUS_INFO_FMT(*gLog, "main begin..."); 57 58 for (int i = 0; i < 5; ++i) 59 { 60 async_accept(); 61 } 62 63 // 捕获信号 64 boost::asio::signal_set signals_(io_svc); 65 signals_.add(SIGINT); 66 signals_.add(SIGTERM); 67 signals_.async_wait([](const boost::system::error_code &ec, int sig) 68 { 69 LOG4CPLUS_INFO_FMT(*gLog, "signal: %d, error_message: %s", 70 sig, ec.message().c_str()); 71 io_svc.stop(); 72 }); 73 74 std::vector<std::thread> vecThread; 75 for (int i = 0; i < 10; ++i) 76 { 77 vecThread.emplace_back(std::thread([](){ 78 LOG4CPLUS_INFO_FMT(*gLog, "thread start..."); 79 io_svc.run(); 80 LOG4CPLUS_INFO_FMT(*gLog, "thread finished."); 81 })); 82 } 83 84 for (size_t i = 0; i < vecThread.size(); ++i) 85 { 86 vecThread[i].join(); 87 } 88 assert(io_svc.stopped(); 89 90 #ifdef WIN32 91 system("pause"); 92 #endif 93 94 return 0; 95 } 96 97 // 标记异步监听，投放到指定io_service 对象中 98 void async_accept() 99 { 100 LOG4CPLUS_INFO_FMT(*gLog, "async_accept waitting..."); 101 102 boost::shared_ptr<boost::asio::ip::tcp::socket> new_sock 103 = boost::make_shared<boost::asio::ip::tcp::socket>(boost::ref(ios_svc)); 104 105 boost::function<void(const boost::system::error_code &> cb_accept; 106 cb_accept = boost::bind(handle_accept, new_sock, _1); 107 acceptor.async_accept(*new_sock, cb_accept); 108 } 109 110 // 监听返回的处理 111 void handle_accept(boost::shared_ptr<boost::asio::ip::tcp::socket> new_conn, 112 const boost::system::error_code &ec) 113 { 114 if (ec != 0) 115 { 116 LOG4CPLUS_INFO(*gLog, "accept failed: " << ec.message()); 117 118 return; 119 } 120 LOG4CPLUS_INFO(*gLog, "a new client connected. " << new_conn->remote_endpoint()); 121 122 async_read_head(new_conn); 123 124 // 处理下一个连接，每次处理完了之后，需要再次accept. 125 // 否则io_service 将只处理一次，然后结束监听。 126 // 所以这里可以处理一个情况，就是当你要结束监听的时候，人要在这里return 127 // 那么io_service 的run() 函数就会stop. 但如果还有其他的异步操作被记录, 128 // run() 函数还是会继续运行，以处理其他的异步操作。 129 async_accept(); 130 } 131 132 // 对一个指定的连接标记异步读头部，然后投放到io_service 对象 133 void async_read_head(boost::shared_ptr<boost::asio::ip::tcp::socket> conn) 134 { 135 // 固定报文头长度为${PACKAGE_LENGTH} 个字节 136 boost::shared_array<char> sa_len(new char[PACKAGE_LENGTH]); 137 138 // 回调函数 139 boost::function<void(const boost::system::error_code &, std::size_t)> cb_msg_len; 140 cb_msg_len = boost::bind(handle_head, conn, sa_len, _1, _2); 141 142 // 异步读，读一个报文的长度，boost::asio::async_read() 函数有个特点， 143 // 它会将这里指定的buffer 缓冲区读满了才会回调handle_head 函数。 144 boost::asio::async_read( 145 *conn, boost::asio::buffer(sa_len.get(), PACKAGE_LENGTH), cb_msg_len); 146 } 147 148 // 头部数据完整读取后的处理函数 149 void handle_head( 150 boost::shared_ptr<boost::asio::ip::tcp::socket> conn, 151 boost::shared_array<char> sa_len, 152 const boost::system::error_code &ec, 153 std::size_t bytes_transfered) 154 { 155 if (!conn->is_open()) 156 { 157 LOG4CPLUS_INFO(*gLog, "socket was not opened."); 158 return ; 159 } 160 161 if (ec != 0) 162 { 163 if (ec == boost::asio::error::eof) 164 { 165 LOG4CPLUS_INFO(*gLog, "Disconnect from " << conn->remote_endpoint()); 166 } 167 else 168 { 169 LOG4CPLUS_INFO(*gLog, "Error on receive: " << ec.message()); 170 } 171 172 return ; 173 } 174 175 // 这里对的数据做处理 176 assert(bytes_transfered == PACKAGE_LENGTH); 177 int32_t len_net = 0; // 网络字节序：数据部分长度 178 int32_t len_loc = 0; // 本地字节序：数据部分长度 179 memcpy(&len_net, sa_len.get(), sizeof(len_net)); 180 len_loc = boost::asio::detail::socket_ops::network_to_host_long(len_net); 181 LOG4CPLUS_INFO_FMT(*gLog, "nLenLoc: %d", len_loc); 182 183 async_read_proto(conn, len_loc); 184 } 185 186 // 对一个指定的连接标记异步读数据部，然后投放到io_service 对象 187 void async_read_proto(boost::shared_ptr<boost::asio::ip::tcp::socket> conn, int32_t len) 188 { 189 // 数据部分 190 boost::shared_array<char> sa_data(new char[len]()); 191 192 // 回调函数 193 boost::function<void(const boost::system::error_code &, std::size_t)> cb_proto; 194 cb_proto = boost::bind(handle_proto, conn, sa_data, _1, _2); 195 196 boost::asio::async_read(*conn, 197 boost::asio::buffer(sa_data.get(), len), cb_proto); 198 } 199 200 // 数据部分读完整后的处理函数 201 void handle_proto( 202 boost::shared_ptr<boost::asio::ip::tcp::socket> conn, 203 boost::shared_array<char> sa_data, 204 const boost::system::error_code &ec, 205 std::size_t bytes_transfered) 206 { 207 if (!conn->is_open()) 208 { 209 LOG4CPLUS_INFO(*gLog, "socket was not opened."); 210 return ; 211 } 212 213 if (ec != 0) 214 { 215 if (ec == boost::asio::error::eof) 216 { 217 LOG4CPLUS_INFO(*gLog, "Disconnect from " << conn->remote_endpoint()); 218 } 219 else 220 { 221 LOG4CPLUS_INFO(*gLog, "Error on receive: " << ec.message()); 222 } 223 return ; 224 } 225 226 // 处理这个proto 数据 227 // 这里将这个数组转换成一个proto, 然后处理这个proto 228 MessageHead pro; 229 if (!pro.ParseFromArray(sa_data.get(), (int32_t)bytes_transfered)) 230 { 231 LOG4CPLUS_ERROR_FMT(*gLog, "ParseFromArray() failed"); 232 return ; 233 } 234 235 int port = conn->remote_endpoint().port(); 236 LOG4CPLUS_INFO_FMT(*gLog, "port: %d\n%s", port, pro.DebugString().c_str()0; 237 238 // 处理完了之后，类似accept 的异步调用一样，需要继续调用异步的读数据 239 // 同样的，如果要结束一个连接，正常的结算应该在这里return 调用。 240 // 当然了，使用socket 的close(), shut_down() 函数也可以关闭这个连接。 241 async_read_head(conn); 242 }
相关文章：