c++ - Boost::asio - 如何中断阻塞的 tcp 服务器线程？

Question

我正在开发一个多线程应用程序，其中一个线程充当从客户端接收命令的 tcp 服务器。线程使用 Boost 套接字和接受器等待客户端连接，从客户端接收命令，将命令传递给应用程序的其余部分，然后再次等待。代码如下:

void ServerThreadFunc()
{
    using boost::asio::ip::tcp;
    boost::asio::io_service io_service;
    tcp::acceptor acceptor(io_service, tcp::endpoint(tcp::v4(), port_no));

    for (;;)
    {
        //  listen for command connection
        tcp::socket socket(io_service);
        acceptor.accept(socket);

        //  connected; receive command
        boost::array<char,256> msg_buf;
        socket.receive(boost::asio::buffer(msg_buf));

        //  do something with received bytes here
    }
}

该线程将大部分时间都花在调用 acceptor.accept() 上。目前，线程仅在应用程序退出时终止。不幸的是，这会在 main() 返回后导致崩溃 - 我相信是因为线程在单例被销毁后尝试访问应用程序的日志记录单例。 (老实说，我来的时候就是这样。)

当应用程序退出时，我怎样才能干净地关闭这个线程？我已经读过原始套接字上的阻塞 accept() 调用可以通过从另一个线程关闭套接字来中断，但这似乎不适用于 Boost 套接字。我尝试使用 Boost asynchronous tcp echo server example 将服务器逻辑转换为异步 i/o ，但这似乎只是将阻塞调用 acceptor::accept() 换成阻塞调用 io_service::run()，所以我只剩下同样的问题:我无法中断的阻塞调用。有什么想法吗？

Answer 1

简而言之，有两种选择:

• 将代码更改为异步(acceptor::async_accept() 和 async_read)，通过 io_service::run() 在事件循环中运行 code>，然后通过 io_service::stop() 取消。
• 通过信号等较低级别的机制强制阻塞调用中断。

我会推荐第一个选项，因为它更便于携带且更易于维护。要理解的重要概念是 io_service::run()只要有待处理的工作就会阻塞。当io_service::stop()被调用时，它将尝试使所有在 io_service::run() 上阻塞的线程尽快返回；它不会中断同步操作，例如 acceptor::accept() 和 socket::receive()，即使在事件循环中调用了同步操作。重要的是要注意 io_service::stop() 是一个非阻塞调用，因此与在 io_service::run() 上阻塞的线程同步必须使用另一个机械师，例如 thread::join() .

这是一个运行 10 秒并监听 8080 端口的示例:

#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/thread.hpp>
#include <iostream>

void StartAccept( boost::asio::ip::tcp::acceptor& );

void ServerThreadFunc( boost::asio::io_service& io_service )
{
  using boost::asio::ip::tcp;
  tcp::acceptor acceptor( io_service, tcp::endpoint( tcp::v4(), 8080 ) );

  // Add a job to start accepting connections.
  StartAccept( acceptor );

  // Process event loop.
  io_service.run();

  std::cout << "Server thread exiting." << std::endl;
}

void HandleAccept( const boost::system::error_code& error,
                   boost::shared_ptr< boost::asio::ip::tcp::socket > socket,
                   boost::asio::ip::tcp::acceptor& acceptor )
{
  // If there was an error, then do not add any more jobs to the service.
  if ( error )
  {
    std::cout << "Error accepting connection: " << error.message() 
              << std::endl;
    return;
  }

  // Otherwise, the socket is good to use.
  std::cout << "Doing things with socket..." << std::endl;

  // Perform async operations on the socket.

  // Done using the socket, so start accepting another connection.  This
  // will add a job to the service, preventing io_service::run() from
  // returning.
  std::cout << "Done using socket, ready for another connection." 
            << std::endl;
  StartAccept( acceptor );
};

void StartAccept( boost::asio::ip::tcp::acceptor& acceptor )
{
  using boost::asio::ip::tcp;
  boost::shared_ptr< tcp::socket > socket(
                                new tcp::socket( acceptor.get_io_service() ) );

  // Add an accept call to the service.  This will prevent io_service::run()
  // from returning.
  std::cout << "Waiting on connection" << std::endl;
  acceptor.async_accept( *socket,
    boost::bind( HandleAccept,
      boost::asio::placeholders::error,
      socket,
      boost::ref( acceptor ) ) );
}

int main()
{
  using boost::asio::ip::tcp;

  // Create io service.
  boost::asio::io_service io_service;

  // Create server thread that will start accepting connections.
  boost::thread server_thread( ServerThreadFunc, boost::ref( io_service ) );

  // Sleep for 10 seconds, then shutdown the server.
  std::cout << "Stopping service in 10 seconds..." << std::endl;
  boost::this_thread::sleep( boost::posix_time::seconds( 10 ) );
  std::cout << "Stopping service now!" << std::endl;

  // Stopping the io_service is a non-blocking call.  The threads that are
  // blocked on io_service::run() will try to return as soon as possible, but
  // they may still be in the middle of a handler.  Thus, perform a join on 
  // the server thread to guarantee a block occurs.
  io_service.stop();

  std::cout << "Waiting on server thread..." << std::endl;
  server_thread.join();
  std::cout << "Done waiting on server thread." << std::endl;

  return 0;
}

在运行时，我打开了两个连接。这是输出:

Stopping service in 10 seconds...
Waiting on connection
Doing things with socket...
Done using socket, ready for another connection.
Waiting on connection
Doing things with socket...
Done using socket, ready for another connection.
Waiting on connection
Stopping service now!
Waiting on server thread...
Server thread exiting.
Done waiting on server thread.