具有Reset功能的多线程同步队列 - 2

// ---------------------------------实现文件---------------------------------//

#include "threadsafequeue.h"

CTreadSafeMsgQueue::CTreadSafeMsgQueue(int QueSize, USHORT InvalidMsgType)

{

INVALID_MSG_TYPE = InvalidMsgType;

MAX_QUE_SIZE = QueSize;

m_Queue = NULL;

m_HeaderToWrite = 0;

m_TailToRead = 0;

m_S_Producer = NULL;

m_S_Consumer = NULL;

m_E_Queue = NULL;

m_WritingThreadNum = 0;

m_ReadingThreadNum = 0;

m_bStop = FALSE;

m_bInitedOK = TRUE; // 注意

if (QueSize > 0) m_Queue = new MsgItem [QueSize];

m_bInitedOK &= (m_Queue != NULL);

if (!m_bInitedOK) return;

m_S_Producer = CreateSemaphore(NULL, MAX_QUE_SIZE, MAX_QUE_SIZE, NULL);

m_bInitedOK &= (m_S_Producer != NULL);

if (!m_bInitedOK) return;

m_S_Consumer = CreateSemaphore(NULL, 0, MAX_QUE_SIZE, NULL);

m_bInitedOK &= (m_S_Consumer != NULL);

if (!m_bInitedOK) return;

m_E_Queue = CreateEvent(NULL, FALSE, TRUE, NULL);

m_bInitedOK &= (m_E_Queue != NULL);

}

CTreadSafeMsgQueue::~CTreadSafeMsgQueue()

{

// 防止新的线程进入

m_bInitedOK = FALSE;

// 如果Reset，等待Reset完成

while (m_bStop) Sleep(SLEEP_TIME);

// 等待至少一类线程（读线程或写线程）退出同步状态

while ((m_WritingThreadNum != 0) && (m_ReadingThreadNum != 0)) Sleep(SLEEP_TIME);

// 此时线程必定阻塞于信号量(Semaphore)状态

// 释放写线程

while (m_WritingThreadNum > 0)

{

// 必须Sleep以出让CPU控制权

ReleaseSemaphore(m_S_Producer, 1, NULL);

Sleep(5);

}

// 释放读线程

if (m_ReadingThreadNum > 0)

{

for (int i = 0; i < MAX_QUE_SIZE; i++)

{

// 将待读取消息类型置为非法类型

m_Queue[m_TailToRead].MsgID = INVALID_MSG_TYPE;

}

while (m_ReadingThreadNum > 0)

{

// 释放读线程

ReleaseSemaphore(m_S_Consumer, 1, NULL);

Sleep(5);

}

}

CloseHandle(m_S_Consumer);

CloseHandle(m_S_Producer);

CloseHandle(m_E_Queue);

delete [] m_Queue;

}