使用 pthreads 作为带队列的线程池？答案

【问题标题】：using pthreads as a thread pool with queue?使用 pthreads 作为带队列的线程池？
【发布时间】：2012-09-03 16:50:28
【问题描述】：

我对 C 很陌生，如果这个问题很离谱，我很抱歉。我使用 Java 执行器服务来制作固定的线程池，但无法理解如何在 C 中做类似的事情。

我学会了如何使用 pthreads 在 c 中创建线程（这看起来很简单），但我不确定如何创建一个固定数量的线程消耗的队列？到目前为止，我所做的所有教程要么自行在主语句中启动线程，要么在 for 循环中执行。如果我采用这种方法，那么当我只需要 3 或 4 个线程时，我将拥有数百万个线程（每个工作项 1 个）并让它们处理一个队列。

这可能吗？如果可以，我需要学习什么？如果 pthreads 无法实现，那么我很乐意使用其他东西，我正在 Mac 上开发并将其部署在 linux 上。

【问题讨论】：

你有什么特别的理由想要这个吗？如今，线程的创建和删除非常有效。除非您的应用程序对此至关重要（很多线程几乎什么都不做），否则您应该专注于其他事情（正确性、活性、种族......）
@JensGustedt 我正在处理一个 100GB 的日志文件，每行需要大约 500 万次操作。我认为数据会比我处理它的速度更快，所以我想要一个阻塞队列并让线程消耗它。 B/C 的工作量如此之大，我认为控制线程数量会更好（不是为了有效地创建/删除线程，而是 B/C 我的处理部分非常占用 CPU）。我在这里做了很多假设（看到我在 Java 程序中所做的事情），所以请纠正我所做的任何错误假设。
您的问题描述听起来很复杂，如果您还没有足够的经验在 C 中正确解决问题，这已经是一个很大的挑战。首先尝试正确解决问题。然后，如果您发现性能不符合您的预期，请进行测量。这是唯一知道的方法。很多时候，即使是有经验的程序员也不能正确估计他们程序的瓶颈所在。线程创建可能有时会成为瓶颈，但它肯定不是首先要寻找的。span>
@JensGustedt 我知道我会在构建和测试方面遵循您的建议（我已经有工作代码，但它没有线程化）。我试图这样做不是为了优化线程方面，而是作为一种通过使用阻塞队列并让线程处理它来控制流量的方式。我正在阅读有关如何创建阻塞队列的信息，但我缺乏了解如何使线程保持活动状态以处理队列（我可以将它们置于无限循环中，但不确定它是否是一个好主意）。只是为了澄清我的意图。我会先从你的方法开始。谢谢。
标准方法是使用生产者消费者模型。这可以通过使用互斥体和条件变量来实现线程之间的信号传递。

标签： c

【解决方案1】：

通过使用条件变量，您可以相当容易地使用单生产者/多消费者模型来做到这一点。请记住，这是一个架构，当然也可以使用其他架构。

在主线程中，你只需创建队列、互斥量和条件变量，然后启动你想运行的线程，伪代码如：

glbQueue = []
glbMutex = new mutex
glbCondVar = new condvar
for i = 1 to 10:
    start thread using thrdFn

下一步是将您需要的任何工作项添加到队列中（使用互斥锁）并根据需要启动条件变量以唤醒线程：

while workitem = getNextWorkItem():
    lock glbMutex
    glbQueue.append (workItem)
    kick glbCondVar
    unlock glbMutex

所有工作项完成后，等待队列清空，然后发布一些哨兵项以关闭线程，然后等待它们完成后再退出。

lock glbMutex
while glbQueue is not empty:
    kick glbCondVar
    unlock glbMutex.
    sleep for a bit
    lock glbMutex
unlock glbMutex.

for i = 1 to 10:
    lock glbMutex
    glbQueue.append (endWorkItem)
    kick glbCondVar
    unlock glbMutex.
    wait for any one thread to exit
exit

完成工作的线程也相对简单。首先，它们在无限循环中运行，等待条件变量被踢出。在该循环中，它们会处理工作项，直到没有更多可用项，然后它们会重新进入睡眠状态。

一旦一个线程接收到结束工作项，它就会退出，保证每个线程得到一个结束项。

换句话说，类似于：

initialise
stillGoing = true
lock glbMutex
while stillGoing:
    wait on glbCondVar using glbMutex
    while stillGoing and glbQueue is not empty:
        extract workItem from glbQueue to thread local storage
        unlock glbMutex.
        if workItem is endWorkItem:
            stillGoing = false
        else:
            do the work specified by workItem
        lock glbMutex
unlock glbMutex
clean up
exit thread

这基本上允许您在队列上拥有固定数量的线程处理项目，并且队列本身受到互斥锁的保护，因此工作线程或主线程之间不会发生争用。

【讨论】：