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

Thread safe FIFO / Queue (Multiple producers, one consumer)

本文关键字：消费者 一个 生产者 安全 FIFO 队列 线程      更新时间：2023-10-16

正如标题所说，我正在尝试编写一个队列，该队列可以由多个线程写入，也可以由单个线程读取。作为一个额外的困难，我需要队列输入保持有序(先进先出(。这就是我迷路的地方。Mutexes不一定按照它们被锁定的顺序被唤醒，所以我不知道我可以用什么来实现我想要的？这里有一个简单的程序来说明我正在尝试做什么：

#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;
}

类Queue的代码就是在那里获得的，请参阅文章底部的完整代码。我所做的只是添加方法"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

正如你所看到的，显然没有订购。谢谢你的帮助！

编辑：我修改了队列，使其为每个表示其顺序的推送调用赋予一个数字，当互斥锁解锁时，队列检查以确保在添加元素之前轮到正确的方法，否则它将返回等待。不确定我是否正确地实现了这一点，但它似乎有效！完整的代码可以在那里找到。

为线程分配要添加的值并期望它们按顺序添加是行不通的，因为你不能强制执行线程的顺序。

相反，让每个线程在运行时添加下一个数字(无论它是什么(。像这样：

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

编辑：增量是原子的还不够。对队列的增量AND推送需要是单个原子操作。这意味着在类之外公开锁变量(它已经是公共的(。

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值之前，它将以错误的顺序添加字母。这是一个很小的窗口，可能不太可能发生，但如果你把计数器增量放在推的同一个锁里，每次都会很完美。

相关文章：