線程安全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 /隊列（多個生產者，一個消費者）

回答

相關問題