【求助】如何使用OpenThreads实现生产消费缓冲模型

xulin_2005

在编写通信模块时需要一个发送缓冲队列，原先在windows下使用win32的信号量可以实现生产消费模型，没有使用OpenThreads，为了跨平台，所以想用OpenThreads实现，但苦于OpenThreads库现有的信号量机制，无法实现，想求教下大侠们，有没有用OpenThreads实现过的。

使用win32的实现代码（标红的部分实现了 写、取、关闭的操作）

1. <blockquote>////////////////////////////////////////////////////      

复制代码

////////////////////////////////////////////////////      

// CPP  

IOData::IOData( int iCountSemaphore /*= 1024*/ )
        {
                _iCountSemaphore = iCountSemaphore;
               
                TCHAR        tsName[1024] = { 0 };
                _stprintf(tsName, TEXT("IOData_Write_%d"), this);
                _hSemaphoreWrite = CreateSemaphore(NULL, 0, _iCountSemaphore, tsName);
                _stprintf(tsName, TEXT("IOData_Read_%d"), this);
                _hSemaphoreRead = CreateSemaphore(NULL, _iCountSemaphore, _iCountSemaphore, tsName);
                _stprintf(tsName, TEXT("IOData_Exit_%d"), this);
                _hExit = CreateEvent(NULL, FALSE, FALSE, tsName);
                ResetEvent(_hExit);
        }


        IOData::~IOData()
        {
                close();
                if ( _hSemaphoreWrite )
                {
                        CloseHandle(_hSemaphoreWrite);
                }
                if ( _hSemaphoreRead )
                {
                        CloseHandle(_hSemaphoreRead);
                }
                if ( _hExit )
                {
                        CloseHandle(_hExit);
                }
        }

        bool IOData::writeData( std::string strData )
        {
                HANDLE hWait[2] = { _hExit, _hSemaphoreRead };
                if ( WAIT_OBJECT_0 == WaitForMultipleObjects(2, hWait, FALSE, INFINITE) )
                {
                       logInfo("向外部发送信息的写入线程关闭");
                        return false;
                }

                _dataQueueMutex.lock();
                _dataQueue.push(strData);
                _dataQueueMutex.unlock();

                ReleaseSemaphore(_hSemaphoreWrite, 1, NULL);

                return true;
        }

        E_IOStatus IOData::readData( std::string & strData )
        {
                HANDLE hWait[2] = { _hExit, _hSemaphoreWrite };
                if ( WAIT_OBJECT_0 == WaitForMultipleObjects(2, hWait, FALSE, INFINITE) )
                {
                      logInfo("向外部发送信息的读取线程关闭");
                        return EIOS_Quit;
                }

                _dataQueueMutex.lock();

                strData = _dataQueue.front();
                _dataQueue.pop();

                _dataQueueMutex.unlock();

                ReleaseSemaphore(_hSemaphoreRead, 1, NULL);


                return EIOS_Normal;
        }

        bool IOData::close()
        {
                if ( !_hExit )
                {
                        return false;
                }
                BOOL bRet = PulseEvent(_hExit);
                if ( bRet )
                {
                        return false;
                }

                return true;
        }
}

foeming

正在学习中!!!

foeming

正在学习中!!!