C++ 並發消息隊列

C++ 並發消息隊列

　　在網上找到了一份POSIX線程顯示的並發消息隊列示例代碼：

      　　http://codereview.stackexchange.com/questions/41604/thread-safe-concurrent-fifo-queue-in-c

　　上面的示例代碼其實是有問題的，他只能對並發Push或者並發Pop進行上鎖，二並不能保證同時Push和Pop是線程安全的，所以在鎖隊列時只能使用一個鎖。同時該代碼並不支持Windows，所以按照這篇文檔的思路想使用標准模板庫（STL）實現一份平台無關的代碼，具體實現如下所示。

#include <queue>
#include <mutex>
#include <thread>
#include <chrono>
#include <memory>
#include <condition_variable>

typedef struct task_tag
{
    int data;
    task_tag( int i ) : data(i) { }
} Task, *PTask;

class MessageQueue
{
public:
    MessageQueue(){}
    ~MessageQueue()
    {
        if ( !m_queue.empty() )
        {
            PTask pRtn = m_queue.front();
            delete pRtn;
        }
        
    }

    void PushTask( PTask pTask )
    {
        std::unique_lock<std::mutex> lock( m_queueMutex );
        m_queue.push( pTask );
        m_cond.notify_one();
    }

    PTask PopTask()
    {
        PTask pRtn = NULL;
        std::unique_lock<std::mutex> lock( m_queueMutex );
        while ( m_queue.empty() )
        {
            m_cond.wait_for( lock, std::chrono::seconds(1) );
        }

        if ( !m_queue.empty() )
        {
            pRtn = m_queue.front();
            if ( pRtn->data != 0 )
                m_queue.pop();
        }

        return pRtn;
    }

private:
    std::mutex m_queueMutex;
    std::condition_variable m_cond; 
    std::queue<PTask> m_queue;
};

void thread_fun( MessageQueue *arguments )
{
    while ( true )
    {
        PTask data = arguments->PopTask();

        if (data != NULL)
        {
            printf( "Thread is: %d\n", std::this_thread::get_id() );
            printf("   %d\n", data->data );
            if ( 0 == data->data ) //Thread end.
                break;
            else
                delete data;
        }
    }

    return;
}

 int main( int argc, char *argv[] )
{
    MessageQueue cq;

    #define THREAD_NUM 3
    std::thread threads[THREAD_NUM];

    for ( int i=0; i<THREAD_NUM; ++i )
        threads[i] = std::thread( thread_fun, &cq );

    int i = 100000;
    while( i > 0 )
    {
        Task *pTask = new Task( --i );
        cq.PushTask( pTask );
    }

    for ( int i=0; i<THREAD_NUM; ++i) 
        threads[i].join();

    //system( "pause" );
    return 0;
}

　　在示例代碼中，我們使主線程向公共隊列cq中Push任務，而其他的線程則負責取出任務並打印任務，由於std::cout並不支持並發線程安全，所以在打印任務時使用printf。主線程new出的任務，在其他線程中使用並銷毀，當主線程發送data為0的任務時，則規定任務發送完畢，而其他的線程獲取到data為0的任務后退出線程，data為0的任務則有消息隊列負責銷毀。整個消息隊列使用標准模板庫實現，現實跨平台。

　　在最初設計std::queue<PTask>的時候，想使用std::queue<std::shared_ptr<Task>>來管理主線程new出來的任務，這樣智能指針則負責處理任務的銷毀工作，但是在多線程並發的時候程序莫名的崩潰，仔細調試了半天，還是沒有找到問題，最終我懷疑智能指針在多線程中是不是有問題呢？所以不得不放棄最初的設計。