造轮子MFC实现BlockingQueue

Posted 2020-09-08 rldts

最近任务需要在MFC下做多线程生产者消费者模式的东西，我找了半天貌似MFC没有类似Java里面BlockingQueue那样的工具（也许是我手残没找到）。

网上好像也有很多大佬去实现这个。但是我没仔细去找，看了看一些资料就想着造个轮子玩玩。

 

实现如下：

主要是利用CCriticalSection保护内置的std::list，然后用CEvent来实现生产者消费者的同步。

 

参考资料：http://stackoverflow.com/questions/6683356/c-templated-producer-consumer-blockingqueue-unbounded-buffer-how-do-i-end-el

 

接口文件：IBlockingQueue.h

 1 #pragma once
 2 
 3 template <class T>
 4 class IBlockingQueue
 5 {
 6 public:
 7     virtual ~IBlockingQueue() {} // 为了让实现这个接口的类的析构函数能被正确调用，参考：http://blog.csdn.net/chaoguodong/article/details/6935524
 8     virtual int size() = 0;
 9     virtual T pop_front() = 0;
10     virtual T pop_back() = 0;
11     virtual void push_front(T val) = 0;
12     virtual void push_back(T val) = 0;
13     virtual bool empty() = 0;
14     virtual void stop() = 0;
15     virtual bool is_stop() = 0;
16 };

 

实现文件：BlockingQueue.h

#pragma once

#include <afxmt.h>
#include <list>

#include "Bridge.h"

// 参考资料：http://stackoverflow.com/questions/6683356/c-templated-producer-consumer-blockingqueue-unbounded-buffer-how-do-i-end-el

class StoppingException
{
public:
    LPCTSTR msg = _T("Stopping");
    StoppingException();
    ~StoppingException();
};

template <class T>
class CBlockingQueue : public IBlockingQueue<T>
{
    CCriticalSection m_cs; // 保护 m_lst
    CEvent m_emptyEvent; // m_lst 为空 就 reset，m_lst 不为空 就 set 放行
    std::list<T> m_lst;
    bool m_stop;
public:
    CBlockingQueue();
    CBlockingQueue(const CBlockingQueue<T>& obj);
    CBlockingQueue<T>& operator=(const CBlockingQueue<T>& obj);
    virtual int size();
    virtual T pop_front();
    virtual T pop_back();
    virtual void push_front(T val);
    virtual void push_back(T val);
    virtual bool empty();
    virtual void stop();
    virtual bool is_stop();
};

template <class T>
CBlockingQueue<T>::CBlockingQueue()
    : m_emptyEvent(FALSE, TRUE, NULL, NULL), m_stop(false) // 初始为RESET，不自动RESET
{}

template <class T>
CBlockingQueue<T>::CBlockingQueue(const CBlockingQueue<T>& obj)
    : m_emptyEvent(FALSE, TRUE, NULL, NULL), m_stop(false) // 初始为RESET，不自动RESET
{
    m_cs.Lock();
    obj.m_cs.Lock();
    m_lst = obj.m_lst;
    obj.m_cs.Unlock();
    m_cs.Unlock();
}

template <class T>
CBlockingQueue<T>& CBlockingQueue<T>::operator = (const CBlockingQueue<T>& obj)
{
    m_cs.Lock();
    obj.m_cs.Lock();
    m_lst = obj.m_lst;
    obj.m_cs.Unlock();
    m_cs.Unlock();
    return *this;
}

template <class T>
int CBlockingQueue<T>::size()
{
    m_cs.Lock();
    int sz = 0;
    sz = m_lst.size();
    m_cs.Unlock();
    return sz;
}
template <class T>
T CBlockingQueue<T>::pop_front()
{
    T val;
    bool done = false;
    while (!done && !is_stop())
    {
        ::WaitForSingleObject(m_emptyEvent.m_hObject, INFINITE);
        if (is_stop())
            throw StoppingException();
        m_cs.Lock();
        if (m_lst.empty())
        {

        }
        else
        {
            val = m_lst.front();
            m_lst.pop_front();
            if (m_lst.empty())
                m_emptyEvent.ResetEvent();
            done = true;
        }
        m_cs.Unlock();

    }
    return val;
}
template <class T>
T CBlockingQueue<T>::pop_back()
{
    T val;
    bool done = false;
    while (!done && !is_stop())
    {
        ::WaitForSingleObject(m_emptyEvent.m_hObject, INFINITE);
        if (is_stop())
            throw StoppingException();
        m_cs.Lock();
        if (m_lst.empty())
        {

        }
        else
        {
            val = m_lst.back();
            m_lst.pop_back();
            if (m_lst.empty())
                m_emptyEvent.ResetEvent();
            done = true;
        }
        m_cs.Unlock();

    }
    return val;
}
template <class T>
void CBlockingQueue<T>::push_front(T val)
{
    m_cs.Lock();
    m_lst.push_front(val);
    m_emptyEvent.SetEvent();
    m_cs.Unlock();
}
template <class T>
void CBlockingQueue<T>::push_back(T val)
{
    m_cs.Lock();
    m_lst.push_back(val);
    m_emptyEvent.SetEvent();
    m_cs.Unlock();
}
template <class T>
bool CBlockingQueue<T>::empty()
{
    m_cs.Lock();
    bool bEmpty = m_lst.empty();
    m_cs.Unlock();
    return bEmpty;
}
template <class T>
bool CBlockingQueue<T>::is_stop()
{
    m_cs.Lock();
    bool bStop = m_stop;
    m_cs.Unlock();
    return bStop;
}
template <class T>
void CBlockingQueue<T>::stop()
{
    m_cs.Lock();
    m_stop = true;
    m_emptyEvent.SetEvent();
    m_cs.Unlock();
}

 

实现文件：BlockingQueue.cpp

#include "BlockingQueue.h"

StoppingException::StoppingException() {}

StoppingException::~StoppingException() {}

 

测试文件：MyApp.h

#pragma once

#include <afxwin.h>

class CMyApp :
    public CWinApp
{
public:
    virtual BOOL InitInstance();
};

 

测试文件：MyApp.cpp

#include "MyApp.h"

#include "BlockingQueue.h"

using namespace std;

class CMainWindow :
    public CFrameWnd
{
public:
    CMainWindow();
    DECLARE_MESSAGE_MAP()
    afx_msg void OnClose();
};

CMainWindow::CMainWindow()
{
    Create(NULL, _T("The Hello Application"), WS_OVERLAPPED | WS_CAPTION |
        WS_SYSMENU | WS_MINIMIZEBOX | WS_THICKFRAME,
        CRect(32, 64, 352, 304));
}

CMyApp myApp;

// 共享的数据‘
IBlockingQueue<int>* pBQ = new CBlockingQueue<int>();

#define NUM_PRODUCER 9 // 生产者个数
#define NUM_CONSUMER 5 // 消费者个数

CWinThread* pThreadProducer[NUM_PRODUCER]; // 生产者线程
CWinThread* pThreadConsumer[NUM_CONSUMER]; // 消费者线程
HANDLE hConsumer[NUM_CONSUMER]; // 消费者HANDLE

// 生产
UINT Produce(LPVOID pParam)
{
    for (int i = 0; i < 10; ++i)
    {
        TRACE(_T("Producer[%d]Producing: %d\n"), ::GetCurrentThreadId(), i);
        pBQ->push_back(i);
        TRACE(_T("Producer[%d]Producing: %d\n"), ::GetCurrentThreadId(), i);
        pBQ->push_front(i);
        TRACE(_T("Producer[%d]Sleeping...\n"), ::GetCurrentThreadId());
        ::Sleep(1000);
    }
    TRACE(_T("Producer[%d]Exiting...\n"), ::GetCurrentThreadId());
    return 0;
}
// 消费
UINT Consume(LPVOID pParam)
{
    try {
        while (true)
        {
            TRACE(_T("Consumer[%d]Waiting...\n"), ::GetCurrentThreadId());
            int val = pBQ->pop_front();
            TRACE(_T("Consumer[%d]Consuming: %d\n"), ::GetCurrentThreadId(), val);
            val = pBQ->pop_back();
            TRACE(_T("Consumer[%d]Consuming: %d\n"), ::GetCurrentThreadId(), val);
        }
    }
    catch (StoppingException& e)
    {
        TRACE(_T("Consumer[%d]%s...\n"), ::GetCurrentThreadId(), e.msg);
    }
    TRACE(_T("Consumer[%d]Exiting...\n"), ::GetCurrentThreadId());
    return 0;
}
// 主线程（UI）
BOOL CMyApp::InitInstance()
{

    _CrtSetBreakAlloc(210);

    m_pMainWnd = new CMainWindow;
    m_pMainWnd->ShowWindow(m_nCmdShow);
    m_pMainWnd->UpdateWindow();

    // 共享的初始数据
    for (int i = 0; i < 100; ++i)
    {
        pBQ->push_back(i);
    }
    // 创建消费者线程
    for (int i = 0; i < NUM_CONSUMER; ++i)
    {
        CWinThread* pThread = ::AfxBeginThread(Consume, NULL, THREAD_PRIORITY_NORMAL, 0, CREATE_SUSPENDED);
        pThread->m_bAutoDelete = FALSE;
        pThreadConsumer[i] = pThread;
        hConsumer[i] = pThread->m_hThread;
    }
    // 启动消费者线程
    for (int i = 0; i < NUM_CONSUMER; ++i)
    {
        pThreadConsumer[i]->ResumeThread();
    }
    // 创建生产者线程
    for (int i = 0; i < NUM_PRODUCER; ++i)
    {
        CWinThread* pThread = ::AfxBeginThread(Produce, NULL, THREAD_PRIORITY_NORMAL, 0, CREATE_SUSPENDED);
        pThread->m_bAutoDelete = FALSE;
        pThreadProducer[i] = pThread;
    }
    // 启动生产者线程
    for (int i = 0; i < NUM_PRODUCER; ++i)
    {
        pThreadProducer[i]->ResumeThread();
    }

    return TRUE;
}

BEGIN_MESSAGE_MAP(CMainWindow, CFrameWnd)
    ON_WM_CLOSE()
END_MESSAGE_MAP()

// 退出主线程
void CMainWindow::OnClose()
{
    pBQ->stop();

    ::WaitForMultipleObjects(NUM_CONSUMER, hConsumer, TRUE, INFINITE);

    for (int i = 0; i < NUM_CONSUMER; ++i)
    {
        delete pThreadConsumer[i];
    }

    for (int i = 0; i < NUM_PRODUCER; ++i)
    {
        delete pThreadProducer[i];
    }

    delete pBQ;

    CFrameWnd::OnClose();
}