boost :: asio多线程异步接受并阻塞读/写服务器

  // Kick off 5 threads
  for (size_t i = 0; i < 5; ++i) {
    boost::thread* t = threads.create_thread(boost::bind(&boost::asio::io_service::run, &io));
    std::cout << \"Creating thread \" << i << \" with id \" << t->get_id() << std::endl;
  }

boost::shared_ptr<CTCPServer> _serverPtr;

void CMyServer::Start()
{      
    //First we must create a few threads
    thread* t = 0;
    for (int i = 0; i < COHConfig::_iThreads; ++i)
    {
        t =_threads.create_thread(bind(&io_service::run, &_io_service));
    }

    //Then we create a server object
    _serverPtr.reset( new CTCPServer(&_io_service, PORT_NUMBER) );    

    //And finally run the server through io_service
    _io_service.post(boost::bind(&CMyServer::RunServer, _serverPtr, &CMyServer::HandleMessage));
}

//This is the function which is called by io_service to start our server
void CMyServer::RunServer(CTCPServer* s, void (*HandleFunction)(shared_ptr<ip::tcp::socket>, deadline_timer*))
{
    s->Run(HandleFunction);
}

//And this is our connection handler
void CMyServer::HandleMessage(shared_ptr< ip::tcp::socket > sockClient, deadline_timer* timer)
{
    cout << \"Handling connection from: \" << sockClient->remote_endpoint().address().to_string() << \":\" << sockClient->remote_endpoint().port() << endl;


    //This is some class which gets socket in its constructor and handles the connection

    scoped_ptr<CMyWorker> myWorker( new CMyWorker(sockClient) );

    msgWorker->Start();

}

//Thanks to this function we can stop our server
void CMyServer::Stop()
{
    _serverPtr->Stop();        
}

#ifndef TCPSERVER_HPP
#define TCPSERVER_HPP

#if defined(_WIN32)
    #define BOOST_THREAD_USE_LIB
#endif

#include <boost/asio.hpp>
#include <boost/noncopyable.hpp>
#include <boost/shared_ptr.hpp>
#include <string>
#include <vector>

class CTCPServer: private boost::noncopyable
{
private:
    bool bKeepRunning;

    boost::asio::io_service* _io_service;
    std::string _sPort;
    boost::asio::ip::tcp::acceptor _acceptor;
    boost::shared_ptr< boost::asio::ip::tcp::socket > _sockClient;
    boost::asio::deadline_timer _timer;
    bool _bIPv6;

    std::string SessionID();
public:

    CTCPServer(boost::asio::io_service* ios, const std::string& sPort, bool bIPv6=false):
        _sPort(sPort),
        _acceptor(*ios),
        _timer(*ios),
        _bIPv6(bIPv6)
    {
        _io_service = ios;
        bKeepRunning = false;
    };
    void Run(void (*HandleFunction)(boost::shared_ptr< boost::asio::ip::tcp::socket > sock, boost::asio::deadline_timer* timer));
    void AsyncAccept(void (*HandleFunction)(boost::shared_ptr< boost::asio::ip::tcp::socket > , boost::asio::deadline_timer* ));
    void AcceptHandler(const boost::system::error_code& e, void (*HandleFunction)(boost::shared_ptr< boost::asio::ip::tcp::socket >, boost::asio::deadline_timer* ));
    void Stop();
    void Stop(void (*StopFunction)());

};

#endif

#include \"TCPServer.hpp\"
#include <boost/thread.hpp>
#include <boost/bind.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/thread/mutex.hpp>
#include <iostream>

using namespace std;

string CTCPServer::SessionID()
{
    ostringstream outs;
    outs << \"[\" << boost::this_thread::get_id() << \"] \";
    return outs.str();
}

void CTCPServer::Run(void (*HandleFunction)(boost::shared_ptr< boost::asio::ip::tcp::socket > , boost::asio::deadline_timer* ))
{
    try
    {
        boost::asio::ip::tcp::resolver resolver(*_io_service);
        boost::asio::ip::tcp::endpoint endpoint;
        if(_bIPv6)
        {
            boost::asio::ip::tcp::resolver::query queryv6(boost::asio::ip::tcp::v6(), _sPort);
            endpoint = *resolver.resolve(queryv6);
        }
        else
        {
            boost::asio::ip::tcp::resolver::query queryv4(boost::asio::ip::tcp::v4(), _sPort);
            endpoint = *resolver.resolve(queryv4);
        }
        _acceptor.open(endpoint.protocol());
        _acceptor.set_option(boost::asio::ip::tcp::acceptor::reuse_address(true));
        _acceptor.set_option(boost::asio::socket_base::enable_connection_aborted(true));
        _acceptor.bind(endpoint);
        _acceptor.listen();
        boost::system::error_code ec;
        bKeepRunning = true;
        AsyncAccept(HandleFunction);
    }
    catch(std::exception& e)
    {
        if(!_bIPv6)
            std::cerr << \"Exception wile creating IPv4 TCP socket on port \"<< _sPort<< \": \" << e.what() << std::endl;
        else
            std::cerr << \"Exception wile creating IPv6 TCP socket on port \"<< _sPort<< \": \" << e.what() << std::endl;
    }
}

void CTCPServer::AsyncAccept(void (*HandleFunction)(boost::shared_ptr< boost::asio::ip::tcp::socket > , boost::asio::deadline_timer* ))
{
    if(bKeepRunning)
    {
        try
        {
            _sockClient.reset(new boost::asio::ip::tcp::socket(*_io_service));
            cout << SessionID() << \"Waiting for connection on port: \" << _sPort << endl;
            _acceptor.async_accept(*_sockClient, boost::bind(&CTCPServer::AcceptHandler, this, boost::asio::placeholders::error, HandleFunction));
        }
        catch(exception& e)
        {
            string sWhat = e.what();
            cout << SessionID() << \"Error while accepting connection: \" << e.what() << endl;
        }
    }
}

void CTCPServer::AcceptHandler(const boost::system::error_code& e,
                            void (*HandleFunction)(boost::shared_ptr< boost::asio::ip::tcp::socket >,
                                                   boost::asio::deadline_timer* ))
{
    if(!e)
    {
        try
        {
            (*_io_service).post(boost::bind(HandleFunction, _sockClient, &_timer));
            AsyncAccept(HandleFunction);
        }
        catch(exception& e)
        {
            cout << SessionID() << \"Exception: \" << e.what() << endl;
        }
    }
}

void CTCPServer::Stop()
{
    cout << SessionID() << \"STOP port \" << _sPort << endl;

    if(!bKeepRunning)
        return;

    bKeepRunning = false;

    try
    {
        _sockClient->close();
    }
    catch(exception& e)
    {
        cout << SessionID() << \"Exception: \" << e.what() << endl;
    }

    try
    {
        _acceptor.cancel();
    }
    catch(exception& e)
    {
        cout << SessionID() << \"Exception: \" << e.what() << endl;
    }

    try
    {
        _acceptor.close();
    }
    catch(exception& e)
    {
        cout << SessionID() << \"Exception: \" << e.what() << endl;
    }
}

void CTCPServer::Stop(void (*StopFunction)())
{
    Stop();
    StopFunction();
}