asio的网络通讯代码练手

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

// MyAsio.cpp: 定义控制台应用程序的入口点。
//

#include "stdafx.h"
#include <iostream>
#include <assert.h>
#include "NetMgr.h"

using namespace DEF;
/*
write by def
技术博客 http://www.cnblogs.com/itdef/ 
技术交流群 群号码：432336863
欢迎c c++ windows驱动爱好者 服务器程序员沟通交流
部分老代码存放地点
http://www.oschina.net/code/list_by_user?id=614253
*/

int main()
{
    NetMgr& p = NetMgr::Instance();
    NetMgr& p1 =NetMgr::Instance();
    assert(&p == &p1);
    //std::cout << &p << " " << &p1;
    p.Start();
    return 0;
}

#pragma once
#include "Pre.h"
#include <iostream>
#include <atomic>

#include <boost/asio.hpp>
#include <boost/bind.hpp>

#include "SyncQueue.h"

/*
write by def
技术博客 http://www.cnblogs.com/itdef/ 
技术交流群 群号码：432336863
欢迎c c++ windows驱动爱好者 服务器程序员沟通交流
部分老代码存放地点
http://www.oschina.net/code/list_by_user?id=614253
*/

NAMESPACEBEGIN(DEF)
using boost::asio::ip::tcp;

const char MAGIC_CHAR = '|';

#pragma  pack (push,1)   
struct BufferHead {
    char flag;    // should equal '|'
    unsigned int bufferLenth;
    BufferHead() {
        flag = 0;
        bufferLenth = 0;
    }
    BufferHead(char f, unsigned int len) {
        flag = f;
        bufferLenth = len;
    }
};

struct BufferStruct {
    enum {
        BUF_MAX_LENTH = 1024 * 2
    };
    BufferHead head;
    char bufBody[BUF_MAX_LENTH];
    BufferStruct() {}
    BufferStruct(const BufferStruct& b) {
        head = b.head;
        memcpy(bufBody, b.bufBody, b.head.bufferLenth);
    }
    char* GetBody() { return bufBody; }
    unsigned int GetLength() { return head.bufferLenth; }
};
#pragma pack(pop)    

class tcp_connection
    : public std::enable_shared_from_this<tcp_connection>
{
public:
    typedef std::shared_ptr<tcp_connection> pointer;
    static pointer create(boost::asio::io_service& io_service, std::shared_ptr<SyncQueue<std::string>> p_messageQueue)
    {
        return pointer(new tcp_connection(io_service, p_messageQueue));
    }
    boost::asio::ip::tcp::socket& socket()
    {
        return socket_;
    }
    void start();
    ~tcp_connection() {
        std::cout << std::endl << "Delete tcp_connection=" << --i << std::endl;
    }
private:
    tcp_connection(boost::asio::io_service& io_service, std::shared_ptr<SyncQueue<std::string>> p_messageQueue)
        : socket_(io_service), p_messageQueue_(p_messageQueue)
    {
        std::cout << "New tcp_connection " << ++i << " times" << std::endl;
    }
    void handle_read_head(const boost::system::error_code& error,size_t bytes_transferred);
    void handle_read_body(const boost::system::error_code& error,size_t bytes_transferred);
    boost::asio::ip::tcp::socket socket_;
    std::shared_ptr<SyncQueue<std::string>>  p_messageQueue_;
    static std::atomic_int  i;
    BufferStruct recvBuff_;
};

class tcp_server
{
public:
    tcp_server(boost::asio::io_service& io_service, int port, std::shared_ptr<SyncQueue<std::string>> p_queue);
private:
    void start_accept() {
        tcp_connection::pointer new_connection =
            tcp_connection::create(acceptor_.get_io_service(), p_messageQueue_);
        acceptor_.async_accept(new_connection->socket(),
            boost::bind(&tcp_server::handle_accept, this, new_connection,
                boost::asio::placeholders::error));
    }
    void handle_accept(tcp_connection::pointer new_connection,
        const boost::system::error_code& error)
    {
        if (!error)
        {
            new_connection->start();
        }

        start_accept();
    }
    std::shared_ptr<SyncQueue<std::string>>  p_messageQueue_;
    boost::asio::io_service& io_service_;
    boost::asio::ip::tcp::acceptor acceptor_;
};



NAMESPACEEND(DEF)

#include "NetAsio.h"

using namespace DEF;
/*
write by def
技术博客 http://www.cnblogs.com/itdef/ 
技术交流群 群号码：432336863
欢迎c c++ windows驱动爱好者 服务器程序员沟通交流
部分老代码存放地点
http://www.oschina.net/code/list_by_user?id=614253
*/
std::atomic_int tcp_connection::i = 0;
tcp_server::tcp_server(boost::asio::io_service& io_service, int port, std::shared_ptr<SyncQueue<std::string>> p_queue):
    io_service_(io_service),
    acceptor_(io_service, boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), port)),
    p_messageQueue_(p_queue)
{
    start_accept();
}

void tcp_connection::start() {
    boost::asio::async_read(socket_, boost::asio::buffer(&recvBuff_.head,sizeof(recvBuff_.head)),
        boost::bind(&tcp_connection::handle_read_head, shared_from_this(),
            boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred));
}


//接受包头内容
void DEF::tcp_connection::handle_read_head(const boost::system::error_code& error,
    size_t bytes_transferred)
{
    if (error == boost::asio::error::eof) {
        return;
    }else if (error) {
        std::cerr << " socket recv head error!! " << error.message() << std::endl;
        return;
    }
    char *pBuf =  recvBuff_.GetBody();
    boost::asio::async_read(socket_, boost::asio::buffer(pBuf, recvBuff_.head.bufferLenth),
        boost::bind(&tcp_connection::handle_read_body, shared_from_this(),
            boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred));
}

void DEF::tcp_connection::handle_read_body(const boost::system::error_code& error,
    size_t bytes_transferred)
{
    //处理接收过程中的错误
    if (error == boost::asio::error::eof) {
        return;
    }else if (error) {
        std::cerr << " socket recv body error!! " << error.message()<<std::endl;
        return;
    }
    //到达此处则接受正常 打印接受内容
    //std::cout << "recv body : " << recvBuff_.GetBody() << std::endl;

    //todo 接受内容放入队列 另行处理 避免网络IO影响处理效率
    assert(p_messageQueue_ != nullptr);
    p_messageQueue_->Put(recvBuff_.GetBody());

    //再次进入异步等待 进行下一次包头的接收
    boost::asio::async_read(socket_, boost::asio::buffer(&recvBuff_.head, sizeof(recvBuff_.head)),
        boost::bind(&tcp_connection::handle_read_head, shared_from_this(),
            boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred));
}

#pragma once
#include "pre.h"
#include <boost/asio.hpp>

#include "SyncQueue.h"
/*
write by def
技术博客 http://www.cnblogs.com/itdef/ 
技术交流群 群号码：432336863
欢迎c c++ windows驱动爱好者 服务器程序员沟通交流
部分老代码存放地点
http://www.oschina.net/code/list_by_user?id=614253
*/
NAMESPACEBEGIN(DEF)
using boost::asio::ip::tcp;

class NetMgr {
public:
    //单例
    static NetMgr&  Instance() {
        static NetMgr instance;
        return instance;
    }
    void Start();
    
private:
    NetMgr() :p_messageQueue_(std::make_shared<SyncQueue<std::string>>(100)) {}
    void NetThread(std::shared_ptr<SyncQueue<std::string>> pQueue);
    void CommandDispatch(std::shared_ptr<SyncQueue<std::string>> pQueue);
    const std::string EXITCOMMAND= "EXIT";
    ~NetMgr() {}
    std::shared_ptr<SyncQueue<std::string>>  p_messageQueue_;
    boost::asio::io_service io_service_;
};






NAMESPACEEND(DEF)

#include <thread>
#include <iostream>
#include "NetMgr.h"
#include "NetAsio.h"

using namespace DEF;
/*
write by def
技术博客 http://www.cnblogs.com/itdef/ 
技术交流群 群号码：432336863
欢迎c c++ windows驱动爱好者 服务器程序员沟通交流
部分老代码存放地点
http://www.oschina.net/code/list_by_user?id=614253
*/
void NetMgr::NetThread(std::shared_ptr<SyncQueue<std::string>> pQueue) {
    assert(pQueue != nullptr);
    boost::asio::io_service& io_s = io_service_;
    tcp_server ts(io_s, 9988, pQueue);
    io_s.run();
}

void NetMgr::CommandDispatch(std::shared_ptr<SyncQueue<std::string>> pQueue) {
    assert(pQueue != nullptr);
    for (int i = 0; i < 11; i++) {
        std::string s;
        pQueue->Take(s);
        std::cout << s << std::endl;
    }

}

void NetMgr::Start() {
    std::thread t1 = std::thread(&NetMgr::NetThread,this, p_messageQueue_);
    std::thread t2 = std::thread(&NetMgr::CommandDispatch,this, p_messageQueue_);
    std::string input;
    std::cout << "Input "" << EXITCOMMAND << "" to exit";
    while (std::cin >> input)
    {
        //todo  add command parse
        if (input == EXITCOMMAND) {
            break;
        }
        std::cout << input << std::endl;
    }
    t1.join();
    t2.join();
}

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

运行如图：