65.9K
CodeProject 正在变化。 阅读更多。
Home

一个标准的、 多线程的动态队列

federico.strati
starIconstarIconstarIconstarIcon
emptyStarIcon
starIcon

4.79/5 (23投票s)

2010年10月6日

GPL3

6分钟阅读

viewsIcon

159972

downloadIcon

5609

这是标准的 Windows/C++ 实现的多线程队列。

引言

这是标准的 Windows/C++ 实现的多线程队列，在

M. Michael and M. Scott. "Nonblocking algorithms and preemption-safe locking on
                          multiprogrammed shared - memory multiprocessors."
Journal of Parallel and Distributed Computing, 51(1):1-26, 1998.

本文实现的队列是在多线程环境中动态队列的实际标准。引用的文章是众多参考资料中的一篇。该队列也在教科书《C++ 并发实战 - 实用多线程》（作者：Anthony Williams）中有详细介绍。

多线程队列的明确目标是允许应用程序中的不同线程以指定对象的格式传递消息。例如，您可以拥有生产者线程，它们可以为消费者线程创建消息，而消费者线程可以响应这些消息并对其执行操作。

本文介绍的队列是根据实时编程环境中对最快执行时间的最高要求所设计的。这是最基础的队列。

算法 - 简要说明

多线程队列基本包含一个初始化部分和两个操作：*push* 和 *pop*，或者称为 *enqueue* 和 *dequeue*。在多线程环境中，重点在于并发入队或出队操作时资源的锁定与解锁。

队列的这两个操作必须允许并发访问共享资源。因此，它们必须使用锁（互斥锁）来协调对共享资源的访问。

这种动态队列的标准实现通过将队列本身表示为一个简单、永不为空的单向链表来解决这个问题；它总是包含至少一个虚拟节点。这使得队列的头部/尾部可以进行并行并发访问，因为总会有一个元素在那里，从而防止对同一节点的并发访问。因此，*push* 和 *pop* 这两个操作只需要各自的锁来协调对队列尾部/头部的并发访问。

这种锁绝不会被同时持有：每个操作都会独立于另一个操作访问队列的头部/尾部。这使得多线程队列的实现速度达到最快，因为每个锁仅用于独立地串行化两个操作；也就是说，每个锁仅用于串行化对队列头部或尾部的多线程访问，而不是串行化对队列头部和尾部自身的*并发*访问。

因此，*push* 和 *pop* 这两个操作可以完全独立运行，并可能并行运行而不会破坏队列的状态。锁仅用于从多个线程安全地单独访问这两个操作中的每一个；也就是说，我们需要串行化对队列头部或尾部的访问。这是使用锁实现多线程队列的最快方法。也可以实现无锁算法，但这需要更高的编程复杂度。

算法 - 伪代码

这是用过程式语言编写的队列伪代码的简要描述。

在与本文相关的代码中，伪代码被翻译成了 C++。因此，初始化将在队列的构造函数中执行。

初始化

structure node_t  {value: data type, next: pointer to node_t}
structure queue_t {Head: pointer to node_t, Tail: 
          pointer to node_t, H_lock: lock type, T_lock: lock type}

INITIALIZE(Q: pointer to queue_t)
node = new node()              # Allocate a free node
node->next = NULL              # Make it the only node in the linked list
Q->Head = Q->Tail = node       # Both Head and Tail point to it
Q->H_lock = Q->T_lock = FREE   # Locks are initially free

入队和出队操作（分别为 push 和 pop）

ENQUEUE or PUSH(Q: pointer to queue_t, value: data type)
node = new node()        # Allocate a new node from the free list
node->value = value      # Copy enqueued value into node
node->next = NULL        # Set next pointer of node to NULL
lock(&Q->T_lock)         # Acquire Tail lock in order to access Tail
   Q->Tail->next = node  # Link node at the end of the linked list
   Q->Tail = node        # Swing Tail to node
unlock(&Q->T_lock)       # Release Tail lock

DEQUEUE or POP(Q: pointer to queue_t, pvalue: pointer to data type): boolean
lock(&Q->H_lock)              # Acquire Head lock in order to access Head
   node = Q->Head             # Read Head
   new_head = node->next      # Read next pointer
   if (new_head == NULL)      # Is queue empty?
      unlock(&Q->H lock)      # Release Head lock before return
      return FALSE            # Queue was empty
   endif
   *pvalue = new_head->value  # Queue not empty. Read value before release
   Q->Head = new_head         # Swing Head to next node
unlock(&Q->H_lock)            # Release Head lock
free(node)                    # Free node
return TRUE                   # Queue was not empty, dequeue succeeded

C++ 中与本文队列相关的解决方案保留了上述伪代码中提出的“移动语义”：指针用于按引用传递和存储构成队列的对象（项）。这与另一种可能的基于模板的方法形成对比，后者会将队列实现为基于模板的队列，其中队列的元素（项）按值存储，即实现“复制语义”。通常，在对执行时间有最高要求的实时编程环境中，“移动语义”是解决此类问题的标准解决方案，因为它本质上比“复制语义”方法更快。因此，本文介绍的队列是使用指针实现的，并遵循此工作源自的原始论文中的解决方案理念。当然，当模板参数本身是指针时，这是一个明显的例外；在这种情况下，我们将为队列恢复“移动语义”。为此目的提供了一个队列的模板版本。

使用代码

多个线程使用 *push* 方法访问队列以入队新项，并且多个线程使用 *pop* 方法访问以检索项。每个对象都通过其指针按引用传递。

让我们声明我们的队列

CMultiThreadSingleQueue Queue;

使用 *push* 方法很简单，就像这样

CObject* ptrToObj = &SomeObject;
Queue.Push( (void*) ptrToObj );

使用 *pop* 方法很简单，就像这样

void* ptrToObj = NULL;
while ( Queue.Pop( ptrToObj ) )
{
  // Do something with the objet, for instance
  if ( ptrToObj )
     ((CObject*) ptrToObj)->DoWork();
}

需要注意的是，*pop* 方法应该在轮询循环中使用，或者与外部事件同步以实现连续操作；例如，在轮询情况下，可以使用以下方法

while (true) 
{
   void* ptrToObj = NULL;
   while ( Queue.Pop( ptrToObj ) )
   {
      // Do something with the objet, for instance
      if ( ptrToObj )
         ((CObject*) ptrToObj)->DoWork();
   }
   Sleep(nTimeInMilliseconds); // Polling wait in the loop
   if ( m_bDoStop ) // boolean to exit thread
       break;
}

而在基于事件同步的操作模式下，可以使用以下方法

while (true) 
{
   // Wait for event signalled
   WaitForSingleObject(hHandleToEvent,nTimeInMilliseconds);
   // Do work
   void* ptrToObj = NULL;
   while ( Queue.Pop( ptrToObj ) )
   {
      // Do something with the objet, for instance
      if ( ptrToObj )
         ((CObject*) ptrToObj)->DoWork();
   }
   if ( m_bDoStop ) // boolean to exit thread
       break;
}

关于演示应用程序

本应用程序演示了在实时编程技术中用于线程间通信的一些标准“队列”的用法。

该示例提供了使用以下内容的示例：

  1. 类 CMultiThreadSingleQueue：具有先入先出 (FIFO) 语义的并发无界队列。此队列旨在用作多个生产者和多个消费者之间的缓冲区。示例显示了 3 个生产者和 2 个消费者的用法。
  2. 类 CCircBuffer：具有先入先出 (FIFO) 语义的并发有界队列。此队列旨在用作单个生产者和单个消费者之间的缓冲区。**不允许**有多个消费者和生产者。消息以 FIFO 顺序在两个线程之间传递，并且也可以存储在共享内存段中用于进程间通信。
  3. 类 CDoubleBuffer：双缓冲翻转结构，**仅**可用于两个线程之间：一个生产者和一个消费者。双缓冲是一种具有两个相同缓冲区的结构，由两个线程以并发方式使用：当一个线程写入一个缓冲区时，另一个线程可以从第二个缓冲区读取；这些操作完成后，缓冲区被交换，过程继续。消息以 FIFO 顺序在两个线程之间传递，并且也可以存储在共享内存段中用于进程间通信。

示例应用程序还演示了用于管理生产者和消费者（工作线程）的线程技术，以及编写简单的多标签对话框应用程序（类 CTabCtrlEx）。在运行演示应用程序之前，必须先编译 GridCtrl 库。这些工作解决方案适用于 Visual Studio 2008 Service Pack 1。

关注点

这是研究中使用的实际标准的带锁多线程队列。

历史

  • 2010 年 10 月 6 日 - 初稿。
  • 2010 年 10 月 7 日 - 添加了关于具体点的更多信息。
  • 2010 年 10 月 21 日 - 添加了演示 Win32/MFC 应用程序。
© . All rights reserved.