hworkqueues.h

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

// HWorkQueues.h: interface for the CHWorkQueues class.
//
//////////////////////////////////////////////////////////////////////

#if !defined(AFX_HWORKQUEUES_H__FA045468_E34F_4210_8741_6BED9572981E__INCLUDED_)
#define AFX_HWORKQUEUES_H__FA045468_E34F_4210_8741_6BED9572981E__INCLUDED_

#if _MSC_VER > 1000
#pragma once
#endif // _MSC_VER > 1000

#include "HWorkQueue.h"

template <class WorkNode>
class CHWorkQueues:public IWorkQueue<WorkNode> 
{
	UNI_DWORD m_QueueCount;
	UNI_DWORD m_ItemCount;
	UNI_DWORD m_DoWithCount;
	HWorkQueue<WorkNode> *m_pQueues;//[MAX_QUEUE];
	CHMutex m_Mutex;
	UNI_BOOL m_bStop;
	void GetMinQueue(int &index,int &min)
	{
		index=0;
		min =-1;
		for(UNI_DWORD i=0;i<m_QueueCount;i++)
		{
			if((UNI_DWORD)m_pQueues[i].GetCount() < (UNI_DWORD)min)
			{
				index=i;
				min=(int)m_pQueues[i].GetCount();
			}
		}
	}

	void DecTotalCount()
	{
		m_Mutex.Lock();
		m_ItemCount--;
		m_Mutex.UnLock();
	}
	
	virtual UNI_BOOL OnThreadInit(){return UNI_TRUE;}
	virtual void OnThreadExit(HWORKLIST &pQueue,CHMutex& mutex){};
	virtual UNI_BOOL IsStop(){return m_bStop;}
protected:
	virtual void OnThreadException(){m_bStop=UNI_TRUE;};
public:
	UNI_ThreadHandle GetThread(int i)
	{
		return m_pQueues[i].GetThread();
	}
	UNI_DWORD  GetItemCount(){return m_ItemCount;}		
	UNI_DWORD  GetDoWithCount(){return m_DoWithCount;}		
	UNI_BOOL Initializ(UNI_DWORD  nQueueCount)
	{
		if(m_pQueues)
		{
			delete []m_pQueues;
			m_pQueues=UNI_NULL;
		}
		if(nQueueCount>0)
			if(m_pQueues=new HWorkQueue<WorkNode>[nQueueCount])
			{
				m_QueueCount=nQueueCount;
				return UNI_TRUE;
			}
		return UNI_FALSE;
	}
	void End(int count=-1)
	{
		m_bStop=UNI_TRUE;
		if(count==-1)
			count=this->m_QueueCount;
		for(int i=0;i<count;i++)
		{
			if(m_pQueues[i].GetThread()!=UNI_NULL)
			{
				m_pQueues[i].WakeThread();
				m_pQueues[i].WaitForEndWork();
				m_pQueues[i].UnInitialize();
			}
		}

	}
	
	BOOL Start()
	{
		m_bStop=UNI_FALSE;
		for(int i=0;i<(int)m_QueueCount;i++)
		{
			if(!this->m_pQueues[i].Initialize(this))
				break;
			if(!this->m_pQueues[i].StartWork())
			{
				this->m_pQueues[i].UnInitialize();
				break;
			}
		}
		if(i!=(int)m_QueueCount)
		{
			this->End(i);
			return FALSE;
		}
		return TRUE;

	};
	
	WorkNode *AddItem(WorkNode &pNode,int &min)
	{
		int index;
		index=-1;
		WorkNode *pTemp=UNI_NULL;
		m_Mutex.Lock();
		GetMinQueue(index,min);
		if(index>=0)
		{
			if(pTemp=this->m_pQueues[index].AddTail(pNode))
			{
				
				this->m_ItemCount ++;
				m_DoWithCount++;
				pTemp->nBadCount=0;
				this->m_pQueues[index].WakeThread();
				
			}
		}
		m_Mutex.UnLock();
		return pTemp;

	}
	
public:
	CHWorkQueues()
	{
		m_bStop=UNI_TRUE;
		m_pQueues=UNI_NULL;
		m_QueueCount=0;
		m_ItemCount=0;
		m_DoWithCount=0;
	};
	virtual ~CHWorkQueues()
	{
		if(m_pQueues)
			delete []m_pQueues;
	};

};

#endif // !defined(AFX_HWORKQUEUES_H__FA045468_E34F_4210_8741_6BED9572981E__INCLUDED_)