Boost线程池

Question

我需要一个线程池来运行我的应用程序，并尽可能地依赖标准的东西（C++11或boost）。我知道有一个非官方的boost线程池类可以解决我的问题，但我宁愿避免使用它，因为它并不在boost库本身中 - 为什么这么多年后它仍然没有成为核心库的一部分呢？

在这个页面和其他地方的一些帖子中，人们建议使用boost :: asio来实现类似于线程池的行为。乍一看，这似乎是我想做的事情，但我发现我所见过的所有实现都没有办法加入当前活动任务，这使得它对我的应用程序毫无用处。为了执行连接操作，它们向所有线程发送停止信号，然后将它们加入。然而，这完全抵消了在线程池中使用的优势，因为这使得新任务需要创建一个新线程。

我想做的事情是：

ThreadPool pool(4);
for (...)
{
    for (int i=0;i<something;i++)
        pool.pushTask(...);
    pool.join();
    // do something with the results
}

有人可以提供一个解决方案吗（除了使用SourceForge上现有的非官方线程池）？ C++11或核心Boost中是否有任何可以帮助我解决问题的东西？

- Dtag

---

你不想使用 boost::threadpool 是因为它是非官方的吗？ - Paul Draper

Answer 1

乍一看，那似乎就是我想做的事情，但我发现所有我见过的实现都没有办法加入当前活动任务，这使得它对我的应用程序毫无用处。为了执行加入操作，它们向所有线程发送停止信号，然后加入它们。然而，在我的使用情况下，这完全抵消了线程池的优势，因为这使得新任务需要创建一个新线程。

我认为你可能误解了asio示例：

如果我没记错（而且已经过了一段时间），在线程池中运行的每个线程都调用了io_service :: run，这意味着实际上每个线程都有一个事件循环和调度器。然后，要让asio完成任务，您可以使用io_service :: post方法将任务发布到io_service，并且asio的调度机制会处理其余部分。只要不调用io_service :: stop，线程池将继续运行，使用您开始运行的尽可能多的线程（假设每个线程都有工作要做或已被分配了io_service :: work对象）。

因此，您不需要为新任务创建新线程，这与线程池的概念相违背。

- Ralf

Answer 2

这似乎是boost::futures 的工作。文档中的示例似乎恰好演示了您想要做的事情。

- Bukes

Answer 3

让每个任务类都从一个具有“OnCompletion(task)”方法/事件的任务派生出来。然后，线程池线程可以在调用任务的主要run()方法之后调用它。

等待单个任务完成就很容易了。OnCompletion()可以执行任何必要的操作来向原始线程发出信号，例如信号量、将任务排队到生产者-消费者队列、调用SendMessage/PostMessage API、Invoke/BeginInvoke等等。

如果原始线程需要等待多个任务全部完成，可以扩展上述方法并向池中发出单个“等待任务”。等待任务具有自己的OnCompletion以通信其他任务的完成情况，并且具有线程安全的“任务计数器”（原子操作或锁），设置为要发出的“主”任务数量。等待任务首先被发出到池中，运行它的线程在等待任务中的私有“allDone”条件变量上等待。然后，将“主”任务发出到池中，并将它们的OnCompletion设置为调用等待任务的方法，该方法将任务计数器逐渐减少至零。当任务计数器达到零时，实现此目标的线程会发出allDone条件变量的信号。然后，等待任务OnCompletion运行，从而表示所有主任务已经完成。

这样的机制不需要持续创建/终止/加入/删除线程池线程，对于发起任务需要如何发出信号没有限制，并且您可以发出尽可能多的此类任务组。但是请注意，每个等待任务都会阻塞一个线程池线程，因此请确保在池中创建一些额外的线程（通常不会有任何问题）。

- Martin James

Answer 4

加入线程意味着等待它停止，如果它停止并且您想要分配一个新任务给它，那么您必须创建一个新的线程。因此，在您的情况下，您应该等待条件（例如boost::condition_variable）指示任务结束。因此，使用这种技术可以很容易地使用boost::asio和boost::condition_variable实现。每个线程调用boost::asio::io_service::run，任务将在不同的线程上安排和执行，最后，每个任务将设置boost::condition_variable或事件递减std::atomic以指示作业结束！这真的很容易，不是吗？

- BigBoss

---

抱歉，当我说“join”时，通常是指等待任务的终止（而不是线程）。但一般来说，我同意条件变量是一种解决方案。然而，在我的外部循环的每次迭代中，现在需要实例化条件变量。我不确定这是否是一个昂贵的操作？至于std::atomic：我不确定父线程如何“等待”原子变量变为零？ - Dtag

我只是以条件变量为例，但你甚至可以使用其他技术，例如使用互斥锁来访问计数变量，当计数达到零时设置条件变量，当它增加时重置它。你已经知道解决方案了！ - BigBoss