线程安全FIFO /队列（多个生产者，一个消费者）

正如标题所说，我正在尝试编写一个队列，可以由多个线程写入并由单个线程读取。作为一个额外的困难，我需要队列输入保持有序（先进先出）。这是我迷失的地方。互斥锁不一定会按照锁定的顺序被唤醒，所以我不知道我可以用什么来实现我想要的功能？下面是一个简单的程序说明我想要做的事：线程安全FIFO /队列（多个生产者，一个消费者）

#include "Queue.h" 
#include <Windows.h> 
#include <fstream> 
#include <mutex> 

using std::ofstream; 

ofstream myFile("result.txt"); 
Queue<int> myQueue; 

DWORD WINAPI WritingThread(LPVOID lpParam); 
DWORD WINAPI LockingThread(LPVOID lpParam); 

int main() 
{ 
    // This thread will block myQueue for 3 seconds 
    CreateThread(NULL, 0, LockingThread, NULL, 0, NULL); 

    // During the locked period, I ask myQueue to push numbers from 0 to 49 
    for (int i = 0; i < 50; i++) 
     CreateThread(NULL, 0, WritingThread, (LPVOID)new int(i), 0, NULL); 

    // If the mutex could wake up in order, myQueue would pop up the numbers in order, but it doesn't. 
    for (int i = 0; i < 50; i++) 
     myFile << myQueue.pop() << ","; 

    return EXIT_SUCCESS; 
} 

DWORD WINAPI LockingThread(LPVOID lpParam) 
{ 
    myQueue.lockQueueFor3Seconds(); 
    return 0; 
} 

DWORD WINAPI WritingThread(LPVOID lpParam) 
{ 
    myQueue.push(*(int*)lpParam); 
    return 0; 
}

该类队列中的代码被送往there, see the bottom of the article for full code.我所做的只是将用于测试目的的方法“lockQueueFor3Seconds”。该方法的定义是这样的：

void lockQueueFor3Seconds() 
{ 
    std::unique_lock<std::mutex> mlock(mutex_); 
    Sleep(3000); 
}

该测试的输出是这样的：

1,43,39,46,36,44,49,40,35,42,32,31,28,41,27,38,24,23,20,34,19,16,15,12,37,11,7,8,3,33,30,0,45,4,26,18,48,21,47,22,25,17,14,10,6,29,9,2,13,5

正如你所看到的，显然不是有序的。谢谢你的帮助！

编辑：我修改了队列，以便它为每个推送调用分配一个数字，以表示它们的顺序，当互斥锁被解锁时，队列检查以确保在添加元素之前它是正确的方法，否则返回等待。不知道我是否正确实施了这个，但它似乎有效！完整的代码可以在there找到。

来源

2016-03-09 MyUsername112358

它们按照您将它们放入队列的顺序进行排序。假设你开始的线程按照你创建的顺序运行，你错了。尝试在每个线程之间创建一个睡眠。 – kfsone

附注：为什么不使用std :: thread？避免'new'，你有内存泄漏。 –

将优先级作为参数传递给线程，并在优先级队列中使用该优先级 –

它永远不会工作分配线程的价值增加，并期望他们为了加入，因为你不能强迫线程的执行顺序。

取而代之的是，让每个线程添加下一个号码（无论它可能）在运行时。像这样：

std::atomic_int counter; 

DWORD WINAPI WritingThread(LPVOID lpParam) 
{ 
    myQueue.push(counter++); 
    return 0; 
}

编辑：增量是原子是不够的。增加和推入队列需要是单个原子操作。这意味着暴露类的外部锁定变量（它已经公开）。

std::atomic_int counter; 

DWORD WINAPI WritingThread(LPVOID lpParam) 
{ 
    unique_lock<mutex> lock(myQueue.m_mutex); 
    myQueue.push(counter++); 
    return 0; 
}

如果您mutex实现让同一个线程调用它多次，将工作。否则，你可以做类似的事情：

void pushAndIncrement(T& item) 
{ 
    std::unique_lock<std::mutex> mlock(mutex_); 
    queue_.push(item); 
      ++item; 
    mlock.unlock(); 
    cond_.notify_one(); 
}

我认为你的解决方案（你说的是工作）仍然有一个竞争条件。如果在字母值递增之后存在上下文切换，但是在push内增加counter值之前，它将按错误的顺序添加该字母。这是一个如此小的窗口，可能不太可能发生，但是如果您将计数器增量放在与推动相同的锁内，那么每次都是完美的。

来源

2016-03-09 02:03:41

我不这么认为。我认为'push'方法自动锁定。 –

我将你的代码复制/粘贴到我的代码上，它仍然返回数字无序，我不知道为什么...你有没有在你身边测试过它？ – MyUsername112358

我以前没有，但是在你说完之后，我创建了一个新项目，添加了你的代码，Queue.h代码，然后运行它。 'results.txt'文件包含从0到49的所有数字。也许尝试做一个完整的重建？只是为了确保您正在运行带有更改的代码。原子应该做的伎俩。 –

线程安全FIFO /队列（多个生产者，一个消费者）

回答

相关问题