asio的网络通讯代码练手

asio的网络基本模板(单例模式消息队列 )

 1 // MyAsio.cpp: 定义控制台应用程序的入口点。
 2 //
 3 
 4 #include "stdafx.h"
 5 #include <iostream>
 6 #include <assert.h>
 7 #include "NetMgr.h"
 8 
 9 using namespace DEF;
10 /*
11 write by def
12 技术博客 http://www.cnblogs.com/itdef/ 
13 技术交流群 群号码：432336863
14 欢迎c c++ windows驱动爱好者 服务器程序员沟通交流
15 部分老代码存放地点
16 http://www.oschina.net/code/list_by_user?id=614253
17 */
18 
19 int main()
20 {
21     NetMgr& p = NetMgr::Instance();
22     NetMgr& p1 =NetMgr::Instance();
23     assert(&p == &p1);
24     //std::cout << &p << " " << &p1;
25     p.Start();
26     return 0;
27 }

MyAsio.cpp

  1 #pragma once
  2 #include "Pre.h"
  3 #include <iostream>
  4 #include <atomic>
  5 
  6 #include <boost/asio.hpp>
  7 #include <boost/bind.hpp>
  8 
  9 #include "SyncQueue.h"
 10 
 11 /*
 12 write by def
 13 技术博客 http://www.cnblogs.com/itdef/ 
 14 技术交流群 群号码：432336863
 15 欢迎c c++ windows驱动爱好者 服务器程序员沟通交流
 16 部分老代码存放地点
 17 http://www.oschina.net/code/list_by_user?id=614253
 18 */
 19 
 20 NAMESPACEBEGIN(DEF)
 21 using boost::asio::ip::tcp;
 22 
 23 const char MAGIC_CHAR = '|';
 24 
 25 #pragma  pack (push,1)   
 26 struct BufferHead {
 27     char flag;    // should equal '|'
 28     unsigned int bufferLenth;
 29     BufferHead() {
 30         flag = 0;
 31         bufferLenth = 0;
 32     }
 33     BufferHead(char f, unsigned int len) {
 34         flag = f;
 35         bufferLenth = len;
 36     }
 37 };
 38 
 39 struct BufferStruct {
 40     enum {
 41         BUF_MAX_LENTH = 1024 * 2
 42     };
 43     BufferHead head;
 44     char bufBody[BUF_MAX_LENTH];
 45     BufferStruct() {}
 46     BufferStruct(const BufferStruct& b) {
 47         head = b.head;
 48         memcpy(bufBody, b.bufBody, b.head.bufferLenth);
 49     }
 50     char* GetBody() { return bufBody; }
 51     unsigned int GetLength() { return head.bufferLenth; }
 52 };
 53 #pragma pack(pop)    
 54 
 55 class tcp_connection
 56     : public std::enable_shared_from_this<tcp_connection>
 57 {
 58 public:
 59     typedef std::shared_ptr<tcp_connection> pointer;
 60     static pointer create(boost::asio::io_service& io_service, std::shared_ptr<SyncQueue<std::string>> p_messageQueue)
 61     {
 62         return pointer(new tcp_connection(io_service, p_messageQueue));
 63     }
 64     boost::asio::ip::tcp::socket& socket()
 65     {
 66         return socket_;
 67     }
 68     void start();
 69     ~tcp_connection() {
 70         std::cout << std::endl << "Delete tcp_connection=" << --i << std::endl;
 71     }
 72 private:
 73     tcp_connection(boost::asio::io_service& io_service, std::shared_ptr<SyncQueue<std::string>> p_messageQueue)
 74         : socket_(io_service), p_messageQueue_(p_messageQueue)
 75     {
 76         std::cout << "New tcp_connection " << ++i << " times" << std::endl;
 77     }
 78     void handle_read_head(const boost::system::error_code& error,size_t bytes_transferred);
 79     void handle_read_body(const boost::system::error_code& error,size_t bytes_transferred);
 80     boost::asio::ip::tcp::socket socket_;
 81     std::shared_ptr<SyncQueue<std::string>>  p_messageQueue_;
 82     static std::atomic_int  i;
 83     BufferStruct recvBuff_;
 84 };
 85 
 86 class tcp_server
 87 {
 88 public:
 89     tcp_server(boost::asio::io_service& io_service, int port, std::shared_ptr<SyncQueue<std::string>> p_queue);
 90 private:
 91     void start_accept() {
 92         tcp_connection::pointer new_connection =
 93             tcp_connection::create(acceptor_.get_io_service(), p_messageQueue_);
 94         acceptor_.async_accept(new_connection->socket(),
 95             boost::bind(&tcp_server::handle_accept, this, new_connection,
 96                 boost::asio::placeholders::error));
 97     }
 98     void handle_accept(tcp_connection::pointer new_connection,
 99         const boost::system::error_code& error)
100     {
101         if (!error)
102         {
103             new_connection->start();
104         }
105 
106         start_accept();
107     }
108     std::shared_ptr<SyncQueue<std::string>>  p_messageQueue_;
109     boost::asio::io_service& io_service_;
110     boost::asio::ip::tcp::acceptor acceptor_;
111 };
112 
113 
114 
115 NAMESPACEEND(DEF)

netasio.h

 1 #include "NetAsio.h"
 2 
 3 using namespace DEF;
 4 /*
 5 write by def
 6 技术博客 http://www.cnblogs.com/itdef/ 
 7 技术交流群 群号码：432336863
 8 欢迎c c++ windows驱动爱好者 服务器程序员沟通交流
 9 部分老代码存放地点
10 http://www.oschina.net/code/list_by_user?id=614253
11 */
12 std::atomic_int tcp_connection::i = 0;
13 tcp_server::tcp_server(boost::asio::io_service& io_service, int port, std::shared_ptr<SyncQueue<std::string>> p_queue):
14     io_service_(io_service),
15     acceptor_(io_service, boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), port)),
16     p_messageQueue_(p_queue)
17 {
18     start_accept();
19 }
20 
21 void tcp_connection::start() {
22     boost::asio::async_read(socket_, boost::asio::buffer(&recvBuff_.head,sizeof(recvBuff_.head)),
23         boost::bind(&tcp_connection::handle_read_head, shared_from_this(),
24             boost::asio::placeholders::error,
25             boost::asio::placeholders::bytes_transferred));
26 }
27 
28 
29 //接受包头内容
30 void DEF::tcp_connection::handle_read_head(const boost::system::error_code& error,
31     size_t bytes_transferred)
32 {
33     if (error == boost::asio::error::eof) {
34         return;
35     }else if (error) {
36         std::cerr << " socket recv head error!! " << error.message() << std::endl;
37         return;
38     }
39     char *pBuf =  recvBuff_.GetBody();
40     boost::asio::async_read(socket_, boost::asio::buffer(pBuf, recvBuff_.head.bufferLenth),
41         boost::bind(&tcp_connection::handle_read_body, shared_from_this(),
42             boost::asio::placeholders::error,
43             boost::asio::placeholders::bytes_transferred));
44 }
45 
46 void DEF::tcp_connection::handle_read_body(const boost::system::error_code& error,
47     size_t bytes_transferred)
48 {
49     //处理接收过程中的错误
50     if (error == boost::asio::error::eof) {
51         return;
52     }else if (error) {
53         std::cerr << " socket recv body error!! " << error.message()<<std::endl;
54         return;
55     }
56     //到达此处则接受正常 打印接受内容
57     //std::cout << "recv body : " << recvBuff_.GetBody() << std::endl;
58 
59     //todo 接受内容放入队列 另行处理 避免网络IO影响处理效率
60     assert(p_messageQueue_ != nullptr);
61     p_messageQueue_->Put(recvBuff_.GetBody());
62 
63     //再次进入异步等待 进行下一次包头的接收
64     boost::asio::async_read(socket_, boost::asio::buffer(&recvBuff_.head, sizeof(recvBuff_.head)),
65         boost::bind(&tcp_connection::handle_read_head, shared_from_this(),
66             boost::asio::placeholders::error,
67             boost::asio::placeholders::bytes_transferred));
68 }

netasio.cpp

 1 #pragma once
 2 #include "pre.h"
 3 #include <boost/asio.hpp>
 4 
 5 #include "SyncQueue.h"
 6 /*
 7 write by def
 8 技术博客 http://www.cnblogs.com/itdef/ 
 9 技术交流群 群号码：432336863
10 欢迎c c++ windows驱动爱好者 服务器程序员沟通交流
11 部分老代码存放地点
12 http://www.oschina.net/code/list_by_user?id=614253
13 */
14 NAMESPACEBEGIN(DEF)
15 using boost::asio::ip::tcp;
16 
17 class NetMgr {
18 public:
19     //单例
20     static NetMgr&  Instance() {
21         static NetMgr instance;
22         return instance;
23     }
24     void Start();
25     
26 private:
27     NetMgr() :p_messageQueue_(std::make_shared<SyncQueue<std::string>>(100)) {}
28     void NetThread(std::shared_ptr<SyncQueue<std::string>> pQueue);
29     void CommandDispatch(std::shared_ptr<SyncQueue<std::string>> pQueue);
30     const std::string EXITCOMMAND= "EXIT";
31     ~NetMgr() {}
32     std::shared_ptr<SyncQueue<std::string>>  p_messageQueue_;
33     boost::asio::io_service io_service_;
34 };
35 
36 
37 
38 
39 
40 
41 NAMESPACEEND(DEF)

netmgr.h

 1 #include <thread>
 2 #include <iostream>
 3 #include "NetMgr.h"
 4 #include "NetAsio.h"
 5 
 6 using namespace DEF;
 7 /*
 8 write by def
 9 技术博客 http://www.cnblogs.com/itdef/ 
10 技术交流群 群号码：432336863
11 欢迎c c++ windows驱动爱好者 服务器程序员沟通交流
12 部分老代码存放地点
13 http://www.oschina.net/code/list_by_user?id=614253
14 */
15 void NetMgr::NetThread(std::shared_ptr<SyncQueue<std::string>> pQueue) {
16     assert(pQueue != nullptr);
17     boost::asio::io_service& io_s = io_service_;
18     tcp_server ts(io_s, 9988, pQueue);
19     io_s.run();
20 }
21 
22 void NetMgr::CommandDispatch(std::shared_ptr<SyncQueue<std::string>> pQueue) {
23     assert(pQueue != nullptr);
24     for (int i = 0; i < 11; i++) {
25         std::string s;
26         pQueue->Take(s);
27         std::cout << s << std::endl;
28     }
29 
30 }
31 
32 void NetMgr::Start() {
33     std::thread t1 = std::thread(&NetMgr::NetThread,this, p_messageQueue_);
34     std::thread t2 = std::thread(&NetMgr::CommandDispatch,this, p_messageQueue_);
35     std::string input;
36     std::cout << "Input \"" << EXITCOMMAND << "\" to exit";
37     while (std::cin >> input)
38     {
39         //todo  add command parse
40         if (input == EXITCOMMAND) {
41             break;
42         }
43         std::cout << input << std::endl;
44     }
45     t1.join();
46     t2.join();
47 }

ntmgr.cpp

如何优雅的退出是个很麻烦的问题，细节在实际工作需要在慢慢打磨

运行如图：

asio的网络通讯代码练手

秒客网

asio的网络通讯代码练手

相关文章