boost asio处理程序中的长时间运行/阻塞操作

Question

现在的情况

我使用boost.asio实现了一个TCP服务器,它目前使用一个io_service对象,我run从单个线程调用该方法.

到目前为止,服务器能够立即回应客户端的请求,因为它在内存中具有所有必要的信息(接收处理程序中不需要长时间运行的操作).

问题

现在需求已经改变,我需要从数据库(使用ODBC)中获取一些信息 - 这基本上是一个长期运行的阻塞操作 - 以便为客户端创建响应.

我看到了几种方法,但我不知道哪种方法最好(可能还有更多方法):

第一种方法

我可以在处理程序中保持长时间运行的操作,只需io_service.run()从多个线程调用即可.我想我会使用尽可能多的线程,因为我有可用的CPU内核？

虽然这种方法很容易实现,但我不认为我会用这种方法获得最佳性能,因为线程数量有限(大部分时间都是空闲的,因为数据库访问更多是I/O绑定操作而不是计算绑定操作).

第二种方法

在本文件的第6部分中,它说:

将线程用于长时间运行的任务

作为单线程设计的一种变体,该设计仍然使用单个io_service :: run()线程来实现协议逻辑.长时间运行或阻塞任务被传递给后台线程,一旦完成,结果将被发布回io_service :: run()线程.

这听起来很有希望,但我不知道如何实现它.任何人都可以为这种方法提供一些代码片段/示例吗？

第三种方法

BorisSchäling在其7.5节中介绍了如何使用自定义服务扩展boost.asio.

这看起来很多工作.与其他方法相比,这种方法有什么好处吗？

Answer 1

这些方法并未明确相互排斥.我经常看到第一个和第二个的组合:

• 一个或多个线程将网络I/O处理为一个io_service.
• 长时间运行或阻止任务将发布到不同的任务中io_service.这io_service用作线程池,不会干扰处理网络I/O的线程.或者,每次需要长时间运行或阻塞任务时,可以生成一个分离的线程; 但是,线程创建/销毁的开销可能会产生明显的影响.

这个答案提供了一个线程池实现.此外,这里有一个基本的例子,试图强调两者之间的相互作用io_services.

#include <iostream>

#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/chrono.hpp>
#include <boost/optional.hpp>
#include <boost/thread.hpp>

/// @brief Background service will function as a thread-pool where
///        long-standing blocking operations may occur without affecting
///        the network event loop.
boost::asio::io_service background_service;

/// @brief The main io_service will handle network operations.
boost::asio::io_service io_service;

boost::optional<boost::asio::io_service::work> work;

/// @brief ODBC blocking operation.
///
/// @brief data Data to use for query.
/// @brief handler Handler to invoke upon completion of operation.
template <typename Handler>
void query_odbc(unsigned int data,
                Handler handler)
{
  std::cout << "in background service, start querying odbc\n";
  std::cout.flush();
  // Mimic busy work.
  boost::this_thread::sleep_for(boost::chrono::seconds(5));

  std::cout << "in background service, posting odbc result to main service\n";
  std::cout.flush();
  io_service.post(boost::bind(handler, data * 2));
}

/// @brief Functions as a continuation for handle_read, that will be
///        invoked with results from ODBC.
void handle_read_odbc(unsigned int result)
{
  std::stringstream stream;
  stream << "in main service, got " << result << " from odbc.\n";
  std::cout << stream.str();
  std::cout.flush();

  // Allow io_service to stop in this example.
  work = boost::none;
}

/// @brief Mocked up read handler that will post work into a background
///        service.
void handle_read(const boost::system::error_code& error,
                 std::size_t bytes_transferred)
{
  std::cout << "in main service, need to query odbc" << std::endl;
  typedef void (*handler_type)(unsigned int);
  background_service.post(boost::bind(&query_odbc<handler_type>,
    21,                // data
    &handle_read_odbc) // handler
  );

  // Keep io_service event loop running in this example.
  work = boost::in_place(boost::ref(io_service));
} 

/// @brief Loop to show concurrency.
void print_loop(unsigned int iteration)
{
  if (!iteration) return;

  std::cout << "  in main service, doing work.\n";
  std::cout.flush();
  boost::this_thread::sleep_for(boost::chrono::seconds(1));
  io_service.post(boost::bind(&print_loop, --iteration));  
}

int main()
{
  boost::optional<boost::asio::io_service::work> background_work(
      boost::in_place(boost::ref(background_service)));

  // Dedicate 3 threads to performing long-standing blocking operations.
  boost::thread_group background_threads;
  for (std::size_t i = 0; i < 3; ++i)
    background_threads.create_thread(
      boost::bind(&boost::asio::io_service::run, &background_service));

  // Post a mocked up 'handle read' handler into the main io_service.
  io_service.post(boost::bind(&handle_read,
    make_error_code(boost::system::errc::success), 0));

  // Post a mockup loop into the io_service to show concurrency.
  io_service.post(boost::bind(&print_loop, 5));  

  // Run the main io_service.
  io_service.run();

  // Cleanup background.
  background_work = boost::none;
  background_threads.join_all();
}

并输出:

in main service, need to query odbc
  in main service, doing work.
in background service, start querying odbc
  in main service, doing work.
  in main service, doing work.
  in main service, doing work.
  in main service, doing work.
in background service, posting odbc result to main service
in main service, got 42 from odbc.

注意,处理主io_service帖的单线程工作background_service,然后继续处理它的事件循环background_service.一旦background_service得到一个结果,它张贴处理程序进入主io_service.