c++ - 具有缓冲能力的专用线程(每个连接一个线程)(c/c++)

Question

我的进程从单个队列中读取需要发送到多个目的地的任务。我们需要保持任务之间的顺序(即 00:00 到达队列的任务需要在 00:01 到达的任务之前发送)因此我们不能使用线程池。需要按目的地维护订单。

一种解决方案是为每个目的地创建一个专用线程。主线程读取队列中的任务并根据目的地找到正确的线程。

这个解决方案有一个问题:如果工作线程很忙，主线程会一直阻塞，导致系统变慢。我需要的是每个线程一个新队列。主线程将资源共享到队列，工作线程读取传入的新队列消息...

我想与 SO 社区分享我的想法，我正在寻找一个接近我描述的 C/C++ 解决方案。有实现这种模型的库吗？

Answer 1

您想要的设计非常简单；我认为您可能可以编写所需的代码并在一两个小时内使其运行。寻找第 3 方库来实现这个可能有点矫枉过正(除非我误解了这个问题)。

特别是，对于每个“工作”线程，您需要一个 FIFO 数据结构(例如 std::queue)、一个 Mutex 和一个“主”线程可以用来向线程发出唤醒和检查信号的机制新消息的数据结构(例如，条件变量、信号量，甚至工作人员在读取时阻塞的套接字对，主服务器可以发送一个字节来唤醒工作人员)。

然后为了向特定的工作线程发送任务，master 会做这样的事情(伪代码):

struct WorkerThreadData & workerThread = _workerThreads[threadIndexIWantToSendTo];
workerThread.m_mutex.Lock();
workerThread.m_incomingTasksList.push_back(theNewTaskObject);
workerThread.m_mutex.Unlock();
workerThread.m_signalMechanism.SignalThreadToWakeUp();  // make sure the worker looks at the task list!

...并且每个工作线程都会有一个这样的事件循环:

struct WorkerThreadData & myData = _workerThreads[myWorkerIndex];
TaskObject * taskObject;
while(1)
{
   myData.m_signalMechanism.WaitForSignal();  // block until the main thread wakes me up

   myData.m_mutex.Lock();
   taskObject = (myData.m_incomingTasks.length() > 0) ? myData.m_incomingTasks.pop_front() : NULL;
   myData.m_mutex.Unlock();

   if (taskObject) 
   {
      taskObject->DoTheWork();
      delete taskObject;
   }
}

这永远不会阻塞主线程(任何时间)，因为 Mutex 只会被任何人短暂持有。特别是，工作线程在处理任务对象时不持有互斥量。