一个可以重用的线程安全生产者消费者队列类

DWORD CFTLThreadTester::ProducerThreadFun(void* pParam) { CFTLThreadTester* pThis = static_cast<CFTLThreadTester*>(pParam); for (LONG index = 0; index < 5; index++) { CMyTestData* pData = new CMyTestData(index,FALSE); FTLTRACE(TEXT("ProducerThread Write pData->Index = %d, CurrentTotalCount = %d\n"), pData->GetCurrentIndex(),CMyTestData::GetTestDataCount()); if (pThis->m_MyTestDataQueue.Append(pData,INFINITE) == ftwtContinue) { } else { delete pData; } pThis->m_ThreadProducer.SleepAndCheckStop(200); } for (LONG index = 6; index < 10; index++) { CMyTestData* pData = new CMyTestData(index,FALSE); FTLTRACE(TEXT("ProducerThread Write pData->Index = %d, CurrentTotalCount = %d\n"), pData->GetCurrentIndex(),CMyTestData::GetTestDataCount()); { CFAutoQueueOperator<CMyTestData*> queueOperator(&pThis->m_MyTestDataQueue,pData,TRUE,INFINITE); if (queueOperator.HaveGotQueue()) //成功获取，并能够加入 { //Donothing } else { delete pData; } } pThis->m_ThreadProducer.SleepAndCheckStop(200); } //传入要求退出的元素 CMyTestData *pQuitData = new CMyTestData(-1,FALSE); if (pThis->m_MyTestDataQueue.Append(pQuitData,INFINITE) != ftwtContinue) { delete pQuitData; } FTLTRACE(TEXT("Now Producer Thread End.\n")); return 0; } DWORD CFTLThreadTester::ResumerThreadFun(void* pParam) { CFTLThreadTester* pThis = static_cast<CFTLThreadTester*>(pParam); CMyTestData* pData = NULL; while (ftwtContinue == pThis->m_MyTestDataQueue.Remove(pData,INFINITE)) { LONG dwCurrentIndex = pData->GetCurrentIndex(); if (dwCurrentIndex == -1) //Quit { delete pData; break; } pThis->m_ThreadResumer.SleepAndCheckStop(500); delete pData; FTLTRACE(TEXT("ResumerThread Read pData->Index = %d, CurrentTotalCount = %d\n"), dwCurrentIndex,CMyTestData::GetTestDataCount()); } { CFAutoQueueOperator<CMyTestData*> queueOperator(&pThis->m_MyTestDataQueue,pData,FALSE,100); } //清除 while (pThis->m_MyTestDataQueue.RemoveAfterStop(pData)) { delete pData; } FTLTRACE(TEXT("Now Resumer Thread End, CurrentTotalCount = %d\n"),CMyTestData::GetTestDataCount()); return 0; } void CFTLThreadTester::test_CFProducerResumerQueue() { m_ThreadProducer.Start(ProducerThreadFun,this); m_ThreadResumer.Start(ResumerThreadFun,this); m_ThreadProducer.Wait(INFINITE,TRUE); m_ThreadResumer.Wait(INFINITE,TRUE); m_ThreadProducer.Start(ProducerThreadFun,this); Sleep(1000); m_ThreadResumer.Start(ResumerThreadFun,this); Sleep(10); m_MyTestDataQueue.Stop(); //停止Queue m_ThreadProducer.Wait(INFINITE,TRUE); m_ThreadResumer.Wait(INFINITE,TRUE); } void CFTLThreadTester::test_CFProducerResumerQueue_ResolveRelease() { //测试保留Slot值 -- CPPUNIT_ASSERT(m_MyTestDataQueue.GetCapability() == MAX_QUEUE_COUNT); CPPUNIT_ASSERT(m_MyTestDataQueue.GetElementCount() == 0); //Queue中没有元素 CPPUNIT_ASSERT(m_MyTestDataQueue.GetReserveSlotCount() == 1); //初始值是1 CPPUNIT_ASSERT(m_MyTestDataQueue.ReserveSlot(1,INFINITE) == ftwtContinue); //可以保留成功 DWORD dwUseSlot = m_MyTestDataQueue.GetCapability() - m_MyTestDataQueue.GetReserveSlotCount(); CPPUNIT_ASSERT(dwUseSlot == 0); //都保留了，可用Slot为0，将不能再加入Element CMyTestData data(999); CPPUNIT_ASSERT(m_MyTestDataQueue.Append(&data,100) == ftwtTimeOut); // CPPUNIT_ASSERT(m_MyTestDataQueue.ReleaseSlot(MAX_QUEUE_COUNT)); //全部释放 CPPUNIT_ASSERT(m_MyTestDataQueue.GetReserveSlotCount() == 0); CPPUNIT_ASSERT(m_MyTestDataQueue.Append(&data,1000) == ftwtContinue); //此时能成功加入 CMyTestData* pTempData = NULL; CPPUNIT_ASSERT(m_MyTestDataQueue.Remove(pTempData,1000) == ftwtContinue); //能成功取出 CPPUNIT_ASSERT(pTempData == &data); //取出的就是之前放入的 CPPUNIT_ASSERT(m_MyTestDataQueue.ReserveSlot(1,1000) == ftwtContinue); //恢复保留Slot为1 }