hworkqueue.h

积下的一点C++编程序库源码

// HWorkQueue.h: interface for the HWorkQueue class.
//
//////////////////////////////////////////////////////////////////////

#if !defined(AFX_HWORKQUEUE_H__EEC22760_2F37_45A1_84BD_3EF72187788E__INCLUDED_)
#define AFX_HWORKQUEUE_H__EEC22760_2F37_45A1_84BD_3EF72187788E__INCLUDED_

#if _MSC_VER > 1000
#pragma once
#endif // _MSC_VER > 1000
#include "HMutex.h"
#include "HList.h"

template <class NodeType>
class IWorkQueue
{
public:
	typedef NodeType HWORKITEM;
	typedef CHList<NodeType> HWORKLIST;
public:
	virtual UNI_BOOL IsStop()=0;
	virtual UNI_BOOL DoWith(NodeType *pItem)=0;
	virtual UNI_BOOL OnThreadInit()=0;
	virtual void OnThreadException()=0;
	virtual void OnThreadExit(CHList<NodeType> &pQueue,CHMutex& mutex)=0;
	virtual void DecTotalCount()=0;
};
template <class NodeType>
class HWorkQueue  
{
	int					m_Count;//任务的个数
	CHList<NodeType>	m_List;//任务队列
	CHMutex				m_Mutex;
	UNI_Semaphore		m_Semaphore;
	IWorkQueue<NodeType> *	m_pWorkQueues;
	UNI_ThreadHandle	m_hThread;

	static UNI_THREADPROC WorkThread(HWorkQueue *pQueue)
	{
		pQueue->m_pWorkQueues->OnThreadInit();
		NodeType*pNode;
		try
		{
			while(!pQueue->m_pWorkQueues->IsStop())
			{
				
				if(pNode=pQueue->GetHeadItem())
				{
					if(pQueue->m_pWorkQueues->DoWith(pNode))
					{
						//DoWith返回真表示处理完成,可以从队列中删除了
						pQueue->Delete(pNode);
						pQueue->m_pWorkQueues->DecTotalCount();
					}
					else
					{
						//DoWith返回假表示处理未完成,放到队列尾部下一次处理.
						pQueue->AddTail(*pNode);
						pQueue->Delete(pNode);
					}
				}
				else
				{
					UNI_WaitSemaphore(pQueue->m_Semaphore);
				}
			}
		}
		catch(...)
		{
			pQueue->m_pWorkQueues->OnThreadException();
		}
		pQueue->m_pWorkQueues->OnThreadExit(pQueue->m_List,pQueue->m_Mutex);
		UNI_ExitThread(0);
	}
	void Delete(NodeType*pItem)
	{
		this->m_Mutex.Lock();
		if(this->m_List.Remove(pItem))
			this->m_Count--;
		this->m_Mutex.UnLock();
	}
	NodeType *GetHeadItem()
	{
		NodeType *pNode;
		this->m_Mutex.Lock();
		pNode= m_List.GetCurNode(m_List.MoveToBegin());
		this->m_Mutex.UnLock();
		return pNode;
	}
public:

	HWorkQueue()
	{
		m_pWorkQueues=UNI_NULL;
		m_Count=0;
		this->m_hThread=(UNI_ThreadHandle)0;

	};
	virtual ~HWorkQueue()
	{
		m_List.EmptyAll();
	};
	UNI_ThreadHandle GetThread(){return m_hThread;}
	int GetCount(){return m_Count;}
	NodeType* AddTail(NodeType &Item)
	{
		NodeType  *pItem;
		this->m_Mutex.Lock();
		if(pItem=this->m_List.AddTail())
		{
			memcpy(pItem,&Item,sizeof(Item));
			this->m_Count++;
		}
		this->m_Mutex.UnLock();
		return pItem;
	}
	UNI_BOOL StartWork()
	{
		UNI_ThreadID dwID;
		if(m_pWorkQueues==UNI_NULL)
			return UNI_FALSE;
		if(UNI_CreateThread((PUNI_THREADPROC)WorkThread,this,&m_hThread,&dwID))
		{
			return UNI_TRUE;
		}
		m_hThread=(UNI_ThreadHandle)0;
		return UNI_FALSE;
	}
	void WaitForEndWork()
	{
		if(m_hThread)
		{
			UNI_WaitCloseThread(m_hThread);
			m_hThread=(UNI_ThreadHandle)0;
		}
	}
	UNI_BOOL Initialize(IWorkQueue<NodeType> *pQueues)
	{
		if(pQueues!=UNI_NULL && UNI_CreateSemaphore(&m_Semaphore,0))
		{
			m_pWorkQueues=pQueues;
			return UNI_TRUE;
		}
		return UNI_FALSE;
	}

	void UnInitialize()
	{
		if(m_pWorkQueues)
		{
			UNI_CloseSemaphore(m_Semaphore);
			m_pWorkQueues=UNI_NULL;
		}
	}
	void WakeThread()
	{
		UNI_ReleaseSemaphore(m_Semaphore);
	}
};

#endif // !defined(AFX_HWORKQUEUE_H__EEC22760_2F37_45A1_84BD_3EF72187788E__INCLUDED_)