基于Win32 SDK实现的一个简易线程池

利用C++实现了一个简易的线程池模型（基于Win32 SDK），方便使用多线程处理任务。共包含Thread.h、Thread.cpp、ThreadPool.h、ThreadPool.cpp四个源文件。功能相对简单，尚待完善。

Thread.h

// Thread.h: interface for the CThread class.
//
//////////////////////////////////////////////////////////////////////

#if !defined(AFX_THREAD_H__9BC45342_F764_4104_B614_1CC8A1964171__INCLUDED_)
#define AFX_THREAD_H__9BC45342_F764_4104_B614_1CC8A1964171__INCLUDED_

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

class CThread  
{
public:
    CThread();
    virtual ~CThread();

public:
    BOOL  IsBusy();
    void  Terminate();
    DWORD GetThreadId() const;
    void  Run(LPTHREAD_START_ROUTINE lpStartAddress, LPVOID lpParameter);

protected:
    void _SetBusy(BOOL bBusy);

private:
    void __Proc();
    static DWORD WINAPI __ThreadProc(LPVOID lpParameter);

protected:
    LPTHREAD_START_ROUTINE m_pThreadProc;
    LPVOID m_pParameter;

    BOOL m_bBusy;
    CRITICAL_SECTION m_csBusy;

    HANDLE m_hEventWait;
    HANDLE m_hEventExit;

    HANDLE m_hThread;
    DWORD  m_dwThreadId;
};

#endif // !defined(AFX_THREAD_H__9BC45342_F764_4104_B614_1CC8A1964171__INCLUDED_)

Thread.cpp

// Thread.cpp: implementation of the CThread class.
//
//////////////////////////////////////////////////////////////////////

#include "StdAfx.h"
#include "Thread.h"


CThread::CThread()
: m_pThreadProc(0)
, m_pParameter(0)
, m_bBusy(0)
, m_hThread(0)
{
    m_hEventWait = CreateEvent(0, TRUE, FALSE, NULL);
    m_hEventExit = CreateEvent(0, TRUE, FALSE, NULL);
    InitializeCriticalSection(&m_csBusy);
    m_hThread = CreateThread(0, 0, __ThreadProc, this, 0, &m_dwThreadId);
}

CThread::~CThread()
{
    Terminate();
    WaitForSingleObject(m_hThread, INFINITE);
    CloseHandle(m_hThread);
    DeleteCriticalSection(&m_csBusy);
    CloseHandle(m_hEventWait);
    CloseHandle(m_hEventExit);
}

void CThread::Terminate()
{
    SetEvent(m_hEventExit);
}

BOOL CThread::IsBusy()
{
    EnterCriticalSection(&m_csBusy);
    BOOL bRet = m_bBusy;
    LeaveCriticalSection(&m_csBusy);
    return bRet;
}

void CThread::_SetBusy(BOOL bBusy)
{
    EnterCriticalSection(&m_csBusy);
    m_bBusy = bBusy;
    LeaveCriticalSection(&m_csBusy);
}

DWORD CThread::GetThreadId() const
{
    return m_dwThreadId;
}

void CThread::Run(LPTHREAD_START_ROUTINE lpStartAddress, LPVOID lpParameter)
{
    if (!IsBusy() && lpStartAddress)
    {
        _SetBusy(TRUE);
        m_pThreadProc = lpStartAddress;
        m_pParameter = lpParameter;
        SetEvent(m_hEventWait);
    }
}

DWORD WINAPI CThread::__ThreadProc(LPVOID lpParameter)
{
    CThread* pThis = (CThread*)lpParameter;
    pThis->__Proc();

    return 0;
}

void CThread::__Proc()
{
    for (;;)
    {
        if (!m_pThreadProc)
        {
            HANDLE hEvents[2] = {m_hEventExit, m_hEventWait};
            DWORD dwRet = WaitForMultipleObjects(2, hEvents, FALSE, INFINITE);
            if (dwRet - WAIT_OBJECT_0 == 0) break;
        }
        m_pThreadProc(m_pParameter);
        m_pThreadProc = NULL;
        m_pParameter = NULL;
        ResetEvent(m_hEventWait);
        _SetBusy(FALSE);
    }
}

ThreadPool.h

// ThreadPool.h: interface for the CThreadPool class.
//
//////////////////////////////////////////////////////////////////////

#if !defined(AFX_THREADPOOL_H__E24F5439_D623_4511_A874_40CC9417DDEE__INCLUDED_)
#define AFX_THREADPOOL_H__E24F5439_D623_4511_A874_40CC9417DDEE__INCLUDED_

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

#pragma warning(push)
#pragma warning(disable: 4786)

#include <queue>
#include <vector>
#include <utility>

#pragma warning(pop)

class CThread;
class CThreadPool  
{
public:
    CThreadPool(size_t nMaxCount);
    virtual ~CThreadPool();

public:
    BOOL IsBusy();
    BOOL Wait(DWORD dwMilliseconds);
    void Run(LPTHREAD_START_ROUTINE lpStartAddress, LPVOID lpParameter);

private:
    void __Proc();
    static DWORD WINAPI __ThreadProc(LPVOID lpParameter);

protected:
    HANDLE m_hThread;
    HANDLE m_hEventExit;

    CRITICAL_SECTION m_csFunc;

    std::vector<CThread*> m_vThreads;
    std::queue< std::pair<LPTHREAD_START_ROUTINE, LPVOID> > m_qFuncs;
};

#endif // !defined(AFX_THREADPOOL_H__E24F5439_D623_4511_A874_40CC9417DDEE__INCLUDED_)

ThreadPool.cpp

// ThreadPool.cpp: implementation of the CThreadPool class.
//
//////////////////////////////////////////////////////////////////////

#include "StdAfx.h"
#include "Thread.h"
#include <algorithm>
#include "ThreadPool.h"


CThreadPool::CThreadPool(size_t nMaxCount)
{
    for (size_t i(0); i < nMaxCount; ++i)
    {
        m_vThreads.push_back(new CThread);
    }
    InitializeCriticalSection(&m_csFunc);
    m_hEventExit = CreateEvent(NULL, TRUE, FALSE, NULL);
    m_hThread = CreateThread(0, 0, __ThreadProc, this, 0, 0);
}

CThreadPool::~CThreadPool()
{
    SetEvent(m_hEventExit);
    WaitForSingleObject(m_hThread, INFINITE);
    CloseHandle(m_hThread);
    for (std::vector<CThread*>::iterator i(m_vThreads.begin())
        ; i != m_vThreads.end(); ++i)
    {
        delete *i;
    }
    m_vThreads.clear();
    DeleteCriticalSection(&m_csFunc);
}

DWORD WINAPI CThreadPool::__ThreadProc(LPVOID lpParameter)
{
    CThreadPool* pThis = (CThreadPool*)lpParameter;
    pThis->__Proc();

    return 0;
}

void CThreadPool::__Proc()
{
    for (;WaitForSingleObject(m_hEventExit, 100) != WAIT_OBJECT_0;)
    {
        EnterCriticalSection(&m_csFunc);
        if (!m_qFuncs.empty())
        {
            for (std::vector<CThread*>::iterator i(m_vThreads.begin())
                ; i != m_vThreads.end(); ++i)
            {
                if (!(*i)->IsBusy() && !m_qFuncs.empty())
                {
                    std::pair<LPTHREAD_START_ROUTINE, LPVOID>& pFunc = m_qFuncs.front();
                    (*i)->Run(pFunc.first, pFunc.second);
                    m_qFuncs.pop();
                }
            }
        }
        LeaveCriticalSection(&m_csFunc);
    }
}

void CThreadPool::Run(LPTHREAD_START_ROUTINE lpStartAddress, LPVOID lpParameter)
{
    EnterCriticalSection(&m_csFunc);
    std::pair<LPTHREAD_START_ROUTINE, LPVOID> pFunc(std::make_pair(lpStartAddress, lpParameter));
    m_qFuncs.push(pFunc);
    LeaveCriticalSection(&m_csFunc);
}

BOOL CThreadPool::IsBusy()
{
    BOOL bRet(FALSE);
    EnterCriticalSection(&m_csFunc);
    if (m_qFuncs.empty())
    {
        for (std::vector<CThread*>::iterator i(m_vThreads.begin())
            ; i != m_vThreads.end(); ++i)
        {
            if ((*i)->IsBusy())
            {
                bRet = TRUE;
                break;
            }
        }
    }
    else bRet = TRUE;
    LeaveCriticalSection(&m_csFunc);
    return bRet;
}

BOOL CThreadPool::Wait(DWORD dwMilliseconds)
{
    DWORD dwTick = GetTickCount();
    for (;IsBusy();)
    {
        Sleep(1);
        if (GetTickCount() - dwTick > dwMilliseconds)
        {
            break;
        }
    }
    return !IsBusy();
}

使用示例：

// Test.cpp : Defines the entry point for the console application.
//

#include "StdAfx.h"
#include "ThreadPool.h"

CRITICAL_SECTION cs;

using namespace std;

DWORD WINAPI ThreadProc(LPVOID lpParameter)
{
    EnterCriticalSection(&cs);
    cout << LPCSTR(lpParameter) << endl;
    LeaveCriticalSection(&cs);
    return 0;
}

int _tmain(int argc, TCHAR* argv[], TCHAR* envp[])
{
    InitializeCriticalSection(&cs);
        
    CThreadPool *pThrealPool = new CThreadPool(2);
    pThrealPool->Run(ThreadProc, "001");
    pThrealPool->Run(ThreadProc, "002");
    pThrealPool->Run(ThreadProc, "003");
    pThrealPool->Run(ThreadProc, "004");
    pThrealPool->Run(ThreadProc, "005");
    pThrealPool->Run(ThreadProc, "006");
    pThrealPool->Run(ThreadProc, "007");
        
    pThrealPool->Wait(INFINIE);
    delete pThrealPool;
        
    DeleteCriticalSection(&cs);

    return nRetCode;
}

 运行结果：

源代码下载：http://t.cn/RvnJe9S