我的想法是創建X個線程,使用KeepRunning方法運行它,該方法有無限循環調用_io_service.run()並在接收到新連接時使用_io_service發送任務async_accept處理程序中的_io_service.poll()。boost :: asio多線程異步接受阻塞讀/寫服務器
我運行服務器與這樣的代碼:
oh::msg::OHServer s("0.0.0.0", "9999", 200);
ConsoleStopServer = boost::bind(&oh::msg::OHServer::Stop, &s);
SetConsoleCtrlHandler(bConsoleHandler, TRUE);
s.Run();
但是當我接收一個連接,然後使用阻塞讀取服務於它在後()方法在MsgWorker類/寫,那麼所有的線程都正在關閉。
我有一個像下面的代碼(這是從http server(服務器)ASIO的一些混合的例子,我的):
OHServer::OHServer(const std::string& sAddress, const std::string& sPort, std::size_t tps)
: _nThreadPoolSize(tps), _acceptor(_io_service), _sockClient(new boost::asio::ip::tcp::socket(_io_service))
{
// Open the acceptor with the option to reuse the address (i.e. SO_REUSEADDR).
boost::asio::ip::tcp::resolver resolver(_io_service);
boost::asio::ip::tcp::resolver::query query(sAddress, sPort);
boost::asio::ip::tcp::endpoint endpoint = *resolver.resolve(query);
_acceptor.open(endpoint.protocol());
_acceptor.set_option(boost::asio::ip::tcp::acceptor::reuse_address(true));
_acceptor.bind(endpoint);
_acceptor.listen();
_acceptor.async_accept(
*_sockClient,
boost::bind(
&OHServer::AcceptConnection,
this,
boost::asio::placeholders::error
)
);
}
void OHServer::KeepRunning()
{
global_stream_lock.lock();
std::cout << "[" << boost::this_thread::get_id()
<< "] Thread Start" << std::endl;
global_stream_lock.unlock();
while(true)
{
try
{
boost::system::error_code ec;
_io_service.run(ec);
if(ec)
{
global_stream_lock.lock();
std::cout << "[" << boost::this_thread::get_id()
<< "] Error: " << ec << std::endl;
global_stream_lock.unlock();
}
break;
}
catch(std::exception & ex)
{
global_stream_lock.lock();
std::cout << "[" << boost::this_thread::get_id()
<< "] Exception: " << ex.what() << std::endl;
global_stream_lock.unlock();
}
}
global_stream_lock.lock();
std::cout << "[" << boost::this_thread::get_id()
<< "] Thread Finish" << std::endl;
global_stream_lock.unlock();
}
void OHServer::Run()
{
// Create a pool of threads to run all of the io_services.
for (std::size_t i = 0; i < _nThreadPoolSize; ++i)
{
boost::shared_ptr<boost::thread> thread(new boost::thread(
boost::bind(&OHServer::KeepRunning, this)));
threads.push_back(thread);
}
cout << "Hit enter to close server" << endl;
cin.get();
}
void OHServer::Stop()
{
boost::system::error_code ec;
_acceptor.close(ec);
_sockClient->shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec);
_sockClient->close(ec);
_io_service.stop();
// Wait for all threads in the pool to exit.
for (std::size_t i = 0; i < threads.size(); ++i)
{
threads[i]->join();
cout << "threads[ "<< i << "]->join();" << endl;
}
}
void OHServer::Post()
{
std::cout << "Accepted new connection." << std::endl;
CMsgWorker *msgWorker = new CMsgWorker(_sockClient);
msgWorker->Start();
delete msgWorker;
}
void OHServer::AcceptConnection(const boost::system::error_code& e)
{
if (!e)
{
_io_service.post(boost::bind(&OHServer::Post, this));
_acceptor.async_accept(
*_sockClient,
boost::bind(
&OHServer::AcceptConnection,
this,
boost::asio::placeholders::error
)
);
}
}
我應該怎麼做線程是仍在等待一些工作從_io_service辦?
感謝您的幫助!
做異步接受操作,但同步讀寫操作有什麼意義? – 2011-05-26 20:19:53