c++ - 使用 GUI 和工作线程提升 Asio 模式

Question

我想使用用于 GUI 的线程和用于某些套接字 IO 的工作线程来实现 Boost Asio 模式。

工作线程将使用boost::asio::io_service 来管理套接字客户端。套接字上的所有操作将仅由工作线程执行。

GUI 线程需要从工作线程发送和接收消息。

我不知道如何使用 Boost Asio 实现这种模式。

我已经以标准的 Asio 方式实现了套接字通信(我从工作线程调用 io_service.run() 并使用 async_read_some/异步发送)。我不需要 strands，因为 io_service.run() 仅从工作线程调用。

现在我正在尝试添加跨线程消息队列。我该如何实现？

我是否也应该从 GUI 线程运行io_service？

或者我应该只使用 strands 和 post 将消息从 GUI 线程发布到工作线程(而不调用 io_service.run() 或来自 GUI 线程的 io_service.poll_one())，并使用操作系统的 GUI 消息循环将消息从工作线程发布到 GUI 线程？

如果我也需要从 GUI 线程调用 io_service.run() 或 io_service.poll_one()，我是否需要使用 strands 在套接字操作上，因为 io_service 在两个线程之间共享？

编辑:为了澄清我的问题，我想尽我所能，使用 Boost Asio 实现消息队列，仅当 Boost Asio 无法完成这项工作时才依赖其他库。

Answer 1

消息传递相当通用。解决问题的方法多种多样，解决方案可能取决于所需的行为细节。例如，阻塞或非阻塞、控制内存分配、上下文等。

• Boost.Lockfree为单个/多个消费者/生产者提供线程安全的无锁非阻塞队列。它往往非常适合事件循环，在这种情况下，阻塞消费者、等待生产者发出同步构造信号并不理想。

  boost::lockfree::queue<message_type> worker_message_queue;
  
  void send_worker_message(const message_type& message)
  {
    // Add message to worker message queue.
    worker_message_queue.push(message);
  
    // Add work to worker_io_service that will process the queue.
    worker_io_service.post(&process_message); 
  }
  
  void process_message()
  {
    message_type message;
  
    // If the message was not retrieved, then return early.
    if (!worker_message_queue.pop(message)) return;
  
    ...
  }
  

• 或者，Boost.Asio的 io_service 可以充当队列。消息只需要绑定(bind)到指定的处理程序即可。

  void send_worker_message(const message_type& message)
  {
    // Add work to worker_io_service that will process the message.
    worker_io_service.post(boost::bind(&process_message, message)); 
  }
  
  void process_message(message_type& message)
  {
    ...
  }
  

---

This评论表明，愿望不仅仅是传递信息。听起来好像最终目标是允许一个线程导致另一个线程调用任意函数。

如果是这种情况，请考虑:

• 使用Boost.Signals2用于托管信号和插槽实现。这允许任意函数注册一个信号。
• 使用 Boost.Asio 的 io_service 设置信号发射。如果 GUI 线程和工作线程都有自己的 io_service，那么工作线程可以将处理程序发布到 GUI 线程的 io_service 中，该处理程序将发出信号。在 GUI 线程的主循环中，它将轮询 io_service，发出信号，并导致从 GUI 线程的上下文中调用插槽。

这是一个完整的示例，其中两个线程将一条消息(作为 unsigned int)相互传递，并导致在另一个线程中调用任意函数。

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

/// @brief io_service dedicated to gui.
boost::asio::io_service gui_service;

/// @brief io_service dedicated to worker.
boost::asio::io_service worker_service;

/// @brief work to keep gui_service from stopping prematurely.
boost::optional<boost::asio::io_service::work> gui_work;

/// @brief hello slot.
void hello(int x)
{
  std::cout << "hello with " << x << " from thread " << 
               boost::this_thread::get_id() << std::endl;
}

/// @brief world slot.
void world(int x)
{
  std::cout << "world with " << x << " from thread " << 
               boost::this_thread::get_id() << std::endl;
}

/// @brief Type for signals.
typedef boost::signals2::signal<void (int)> signal_type;

void emit_then_notify_gui(signal_type& signal, unsigned int x);

/// @brief Emit signals then message worker.
void emit_then_notify_worker(signal_type& signal, unsigned int x)
{
  // Emit signal, causing registered slots to run within this thread.
  signal(x);

  // If x has been exhausted, then cause gui service to run out of work.
  if (!x)
  {
    gui_work = boost::none;
  }
  // Otherwise, post work into worker service.
  else
  {
    std::cout << "GUI thread: " << boost::this_thread::get_id() << 
                 " scheduling other thread to emit signals" << std::endl;
    worker_service.post(boost::bind(
        &emit_then_notify_gui,
        boost::ref(signal), --x));
  }  
}

/// @brief Emit signals then message worker.
void emit_then_notify_gui(signal_type& signal, unsigned int x)
{
  // Emit signal, causing registered slots to run within this thread.
  signal(x);

  // If x has been exhausted, then cause gui service to run out of work.
  if (!x)
  {
    gui_work = boost::none;
  }
  // Otherwise, post more work into gui.
  else
  {
    std::cout << "Worker thread: " << boost::this_thread::get_id() << 
                 " scheduling other thread to emit signals" << std::endl;
    gui_service.post(boost::bind(
        &emit_then_notify_worker,
        boost::ref(signal), --x));
  }  
}

void worker_main()
{
  std::cout << "Worker thread: " << boost::this_thread::get_id() << std::endl;
  worker_service.run();
}

int main()
{
  signal_type signal;

  // Connect slots to signal.
  signal.connect(&hello);
  signal.connect(&world);

  boost::optional<boost::asio::io_service::work> worker_work(
     boost::ref(worker_service));
  gui_work = boost::in_place(boost::ref(gui_service));

  std::cout << "GUI thread: " << boost::this_thread::get_id() << std::endl;

  // Spawn off worker thread.
  boost::thread worker_thread(&worker_main);

  // Add work to worker.
  worker_service.post(boost::bind(
      &emit_then_notify_gui,
      boost::ref(signal), 3));

  // Mocked up GUI main loop.
  while (!gui_service.stopped())
  {
    // Do other GUI actions.

    // Perform message processing.
    gui_service.poll_one();
  }

  // Cleanup.
  worker_work = boost::none;
  worker_thread.join();
}

及其输出:

GUI thread: b7f2f6d0Worker thread: b7f2eb90hello with 3 from thread b7f2eb90world with 3 from thread b7f2eb90Worker thread: b7f2eb90 scheduling other thread to emit signalshello with 2 from thread b7f2f6d0world with 2 from thread b7f2f6d0GUI thread: b7f2f6d0 scheduling other thread to emit signalshello with 1 from thread b7f2eb90world with 1 from thread b7f2eb90Worker thread: b7f2eb90 scheduling other thread to emit signalshello with 0 from thread b7f2f6d0world with 0 from thread b7f2f6d0