前言

一、线程池的设计原理

1.1 概念

1.2 优缺点

二、线程池的实现

2.1 业务流程图

2.2 类的介绍

2.3 类间关系图

2.4 类的实现

2.4.1 IThreadPool类

2.4.2 ITask类

2.4.3 CMyMutex类

2.4.4 CMyThreadList类

2.4.5 CMyThreadStack类

2.4.6 CMyTaskQueue类

2.4.7 CTaskThread类

2.4.8 CThreadPool类

三、测试

3.1 demo源码

3.2 demo演示结果

总结

前言

曾经接触过一个边扫描边识别图像的小项目，对时间和计算机资源要求比较高，针对这种情况，运用了C++线程池来快速、批量处理图像。

一、线程池的设计原理

1.1 概念

线程池是大量线程的集合，里存放了很多可复用的辅助线程，用于批量处理任务。线程池是一种生产者-消费者模式（通过一系列容器来解决生产者和消费者的强耦合问题），生产者把任务丢给线程池，线程池动态分配线程并处理任务。

1.2 优缺点

线程池的优点：

降低线程创建及销毁带来的资源消耗；

避免线程间互抢资源而导致阻塞现象；

提高线程的可管理性和稳定性；

线程池的缺点：

占用一定量的内存空间；

CPU调度开销随着线程数量增加；

程序实现的复杂度变高；

二、线程池的实现

2.1 业务流程图

2.2 类的介绍

名称	描述
IThreadPool	线程池接口
ITask	任务类接口
CMyMutex	互斥类
CMyThreadList	活动线程列表类
CMyThreadStack	空闲线程堆栈类
CMyTaskQueue	任务队列类
CTaskThread	任务处理线程类
CThreadPool	线程池管理类

2.3 类间关系图

2.4 类的实现

2.4.1 IThreadPool类

#pragma once
#include "TaskThread.h"// Thread Pool Interface
class IThreadPool
{
public:virtual bool RecycleThread(CTaskThread *pTaskThread) = 0;virtual void SetDecodeResult(DWORD dwResult) = 0;
};

2.4.2 ITask类

#pragma once// Task Interface
class ITask {
public:virtual unsigned long TaskProc() = 0;
};

2.4.3 CMyMutex类

#pragma once#include <Windows.h>class CMyMutex {
public:CMyMutex(void) {::InitializeCriticalSection(&m_csMutex);}virtual ~CMyMutex(void) {::DeleteCriticalSection(&m_csMutex);}bool Lock() {::EnterCriticalSection(&m_csMutex);return true;}bool Unlock() {::LeaveCriticalSection(&m_csMutex);return true;}private:CRITICAL_SECTION m_csMutex;
};

2.4.4 CMyThreadList类

#pragma once
#include <list>
#include "MyMutex.h"
#include "TaskThread.h"class CMyThreadList {
public:CMyThreadList(void) {}virtual ~CMyThreadList(void) {}bool Add(CTaskThread *pTaskThread) {m_mtxSync.Lock();bool bResult = false;if (pTaskThread) {m_lstTaskThread.push_back(pTaskThread);bResult = true;}m_mtxSync.Unlock();return bResult;}bool Remove(CTaskThread *pTaskThread) {m_mtxSync.Lock();bool bResult = false;if (!m_lstTaskThread.empty()) {m_lstTaskThread.remove(pTaskThread);bResult = true;}m_mtxSync.Unlock();return bResult;}int Size() {m_mtxSync.Lock();int nStackSize = m_lstTaskThread.size();m_mtxSync.Unlock();return nStackSize;}bool IsEmpty() {m_mtxSync.Lock();bool bEmpty = m_lstTaskThread.empty();m_mtxSync.Unlock();return bEmpty;}bool Clear() {m_mtxSync.Lock();while (!m_lstTaskThread.empty()) {}bool bEmpty = (0 == m_lstTaskThread.size());m_mtxSync.Unlock();return bEmpty;}private:CMyMutex m_mtxSync;std::list<CTaskThread *> m_lstTaskThread;
};

2.4.5 CMyThreadStack类

#pragma once
#include <stack>
#include "MyMutex.h"
#include "TaskThread.h"class CMyThreadStack
{
public:CMyThreadStack(void) {}virtual ~CMyThreadStack(void) {}CTaskThread *Pop() {m_mtxSync.Lock();CTaskThread *pTaskThread = NULL;if (!m_stkTaskThread.empty()) {pTaskThread = m_stkTaskThread.top();m_stkTaskThread.pop();}m_mtxSync.Unlock();return pTaskThread;}bool Push(CTaskThread *pTaskThread) {m_mtxSync.Lock();bool bResult = false;if (NULL != pTaskThread) {m_stkTaskThread.push(pTaskThread);bResult = true;}m_mtxSync.Unlock();return bResult;}int Size() {m_mtxSync.Lock();int nStackSize = m_stkTaskThread.size();m_mtxSync.Unlock();return nStackSize;}bool IsEmpty() {m_mtxSync.Lock();bool bEmpty = m_stkTaskThread.empty();m_mtxSync.Unlock();return bEmpty;}bool Clear() {m_mtxSync.Lock();while (!m_stkTaskThread.empty()) {CTaskThread *pTaskThread = m_stkTaskThread.top();m_stkTaskThread.pop();pTaskThread->SuspendThread();delete pTaskThread;}bool bEmpty = (0 == m_stkTaskThread.size());m_mtxSync.Unlock();return bEmpty;}private:CMyMutex m_mtxSync;std::stack<CTaskThread *> m_stkTaskThread;
};

2.4.6 CMyTaskQueue类

#pragma once
#include <queue>
#include "ITask.h"
#include "MyMutex.h"class CMyTaskQueue {
public:CMyTaskQueue(void) {}virtual ~CMyTaskQueue(void) {}ITask *Pop() {m_mtxSync.Lock();ITask *pMyTask = NULL;if (!m_queMyTask.empty()) {pMyTask = m_queMyTask.front();m_queMyTask.pop_front();}m_mtxSync.Lock();return pMyTask;}bool PushFront(ITask *pMyTask) {m_mtxSync.Lock();bool bResult = false;if (pMyTask) {m_queMyTask.push_front(pMyTask);bResult = true;}m_mtxSync.Unlock();return bResult;}bool PushBack(ITask *pMyTask) {m_mtxSync.Lock();bool bResult = false;if (pMyTask) {m_queMyTask.push_back(pMyTask);bResult = true;}m_mtxSync.Unlock();return bResult;}int Size() {m_mtxSync.Lock();int nStackSize = m_queMyTask.size();m_mtxSync.Unlock();return nStackSize;}bool IsEmpty() {m_mtxSync.Lock();bool bEmpty = m_queMyTask.empty();m_mtxSync.Unlock();return bEmpty;}bool Clear() {m_mtxSync.Lock();while (!m_queMyTask.empty()) {}bool bEmpty = (0 == m_queMyTask.size());m_mtxSync.Unlock();return bEmpty;}private:CMyMutex m_mtxSync;std::deque<ITask *> m_queMyTask;
};

2.4.7 CTaskThread类

头文件

#pragma once
#include <Windows.h>class ITask;
class IThreadPool;class CTaskThread
{
public:CTaskThread(IThreadPool *pIThreadPool);virtual ~CTaskThread(void);// Create & start a threadbool StartThread();// Startup the threadbool ResumeThread();// Stop the threadbool SuspendThread();// Assign a task to this threadbool AssignTask(ITask *pMyTask);// Thread default handlerstatic DWORD WINAPI ThreadProc(LPVOID lParam);private:BOOL    m_bExitThread;DWORD m_dwThreadID;HANDLE m_hThread;HANDLE    m_hEvent;ITask      *m_pMyTask;IThreadPool  *m_pIThreadPool;
};

源文件

#include "TaskThread.h"
#include "ThreadPool.h"CTaskThread::CTaskThread(IThreadPool *pIThreadPool)
{m_bExitThread = FALSE;m_hThread = NULL;m_hEvent = ::CreateEvent(NULL, FALSE, FALSE, NULL);m_pMyTask = NULL;m_pIThreadPool = pIThreadPool;
}CTaskThread::~CTaskThread(void)
{m_bExitThread = TRUE;if (m_hThread) {DWORD dwResult = ::WaitForSingleObject(m_hThread, 100);if (WAIT_OBJECT_0 != dwResult) {TerminateThread(m_hThread, 0);}CloseHandle(m_hThread);m_hThread = NULL;}if (m_hEvent) {CloseHandle(m_hEvent);m_hEvent = NULL;}
}bool CTaskThread::StartThread()
{if (!m_hThread) {m_hThread = ::CreateThread(NULL, 0, ThreadProc, this, 0, &m_dwThreadID);if (INVALID_HANDLE_VALUE == m_hThread) {return false;}printf("New a thread: %d\n", m_dwThreadID);}return true;
}bool CTaskThread::ResumeThread()
{::SetEvent(m_hEvent);m_bExitThread = FALSE;return true;
}bool CTaskThread::SuspendThread()
{::ResetEvent(m_hEvent);m_bExitThread = TRUE;return true;
}bool CTaskThread::AssignTask(ITask *pMyTask)
{if (!pMyTask) return false;m_pMyTask = pMyTask;return true;
}DWORD WINAPI CTaskThread::ThreadProc(LPVOID lParam)
{CTaskThread *pThis = (CTaskThread*)lParam;while (!pThis->m_bExitThread) {DWORD dwResult = ::WaitForSingleObject(pThis->m_hEvent, INFINITE);if (WAIT_OBJECT_0 == dwResult) {if (pThis->m_pMyTask) {printf("Current thread: %d, Current task: %d\n", (int)pThis->m_dwThreadID, (int)pThis->m_pMyTask);// Task is validDWORD dwResult = pThis->m_pMyTask->TaskProc();pThis->m_pIThreadPool->SetDecodeResult(dwResult);delete pThis->m_pMyTask;pThis->m_pMyTask = NULL;// Recycle this threadpThis->m_pIThreadPool->RecycleThread(pThis);}}}return 0;
}

2.4.8 CThreadPool类

头文件

#pragma once
#include "ITask.h"
#include "IThreadPool.h"
#include "MyThreadStack.h"
#include "MyThreadList.h"
#include "MyTaskQueue.h"typedef void(*DecodeResult)(char *pResultBuf, unsigned int unResultBufSize);enum EUM_PRIORITY
{NORMAL,HIGH
};class CThreadPool : public IThreadPool
{public:// Use singleton mode to create objectsstatic CThreadPool* GetInstance(void);// Default destructorvirtual ~CThreadPool(void);// Push an idle thread into stackvirtual bool PushIdleThread(CTaskThread *pTaskThread);// Pop an idle thread from stackvirtual CTaskThread* PopIdleThread();virtual bool RecycleThread(CTaskThread *pTaskThread);// Assign a taskvirtual bool SetTask(ITask *pMyTask, EUM_PRIORITY eumPriority= NORMAL);// Get a taskvirtual ITask* GetTask();// Set callback functionbool SetCallbackFun(void *pCallback);// Get the result of image decodingvoid SetDecodeResult(DWORD dwResult);private:// Default constructorCThreadPool(void);// Perform initializationbool _Initialize(int nThreadAmount);// Perform cleanupvoid _Cleanup();// Dynamically allocate threadsbool _ReallocateThreads();DecodeResult m_fDecodeResult;private:int m_nThreadActiveLow;int m_nThreadActiveHigh;int m_nThreadTotal;int m_nCpuCoreTotal;CMyMutex        *m_pMyMutex;CMyTaskQueue    *m_pTaskQueue;CMyThreadStack    *m_pIdleThreadStack;CMyThreadList   *m_pActiveThreadList;
};

源文件

#include "ThreadPool.h"
#include "TaskThread.h"
#include <cassert>
#include<stdlib.h>CThreadPool* CThreadPool::GetInstance(void)
{static CThreadPool* pNewThreadPool = NULL;if (NULL == pNewThreadPool) {pNewThreadPool = new CThreadPool();assert(pNewThreadPool);}return pNewThreadPool;
}CThreadPool::CThreadPool(void)
{m_pMyMutex = NULL;m_pTaskQueue = NULL;m_pIdleThreadStack = NULL;m_pActiveThreadList = NULL;m_fDecodeResult = NULL;SYSTEM_INFO SysemInfo;GetSystemInfo(&SysemInfo);int nCpuCount = SysemInfo.dwNumberOfProcessors;nCpuCount = 4;m_nThreadActiveLow = 4;m_nThreadActiveHigh = 10;m_nThreadTotal = 10;m_nCpuCoreTotal = nCpuCount;// Initialize thread pool_Initialize(m_nThreadTotal);
}CThreadPool::~CThreadPool(void)
{// Clearup thread pool_Cleanup();
}bool CThreadPool::PushIdleThread(CTaskThread *pTaskThread)
{bool bResult = false;if (m_pIdleThreadStack) {m_pIdleThreadStack->Push(pTaskThread);bResult = true;}return bResult;
}CTaskThread* CThreadPool::PopIdleThread()
{CTaskThread *pTaskThread = NULL;if (!m_pIdleThreadStack->IsEmpty()) {pTaskThread = m_pIdleThreadStack->Pop();}return pTaskThread;
}bool CThreadPool::RecycleThread(CTaskThread *pTaskThread)
{if (!m_pTaskQueue->IsEmpty()) {ITask *pCurTask = m_pTaskQueue->Pop();if (pCurTask && pTaskThread) {pTaskThread->AssignTask(pCurTask);pTaskThread->ResumeThread();}// Dynamically allocate threads_ReallocateThreads();}else {m_pActiveThreadList->Remove(pTaskThread);m_pIdleThreadStack->Push(pTaskThread);}return true;
}bool CThreadPool::SetTask(ITask *pMyTask, EUM_PRIORITY eumPriority)
{if (HIGH == eumPriority) {m_pTaskQueue->PushFront(pMyTask);}else {m_pTaskQueue->PushBack(pMyTask);}if (!m_pTaskQueue->IsEmpty()) {ITask *pCurTask = m_pTaskQueue->Pop();CTaskThread *pTaskThread = m_pIdleThreadStack->Pop();if (pCurTask && pTaskThread) {pTaskThread->AssignTask(pCurTask);pTaskThread->ResumeThread();// Dynamically allocate threads_ReallocateThreads();return true;}}return false;
}ITask* CThreadPool::GetTask()
{ITask *pMyTask = NULL;if (!m_pTaskQueue->IsEmpty()) {pMyTask = m_pTaskQueue->Pop();}return pMyTask;
}bool CThreadPool::_Initialize(int nThreadAmount)
{if (!m_pMyMutex) m_pMyMutex = new CMyMutex();if (!m_pTaskQueue) m_pTaskQueue = new CMyTaskQueue();if (!m_pIdleThreadStack) m_pIdleThreadStack = new CMyThreadStack();if (!m_pActiveThreadList) m_pActiveThreadList = new CMyThreadList();assert(m_pMyMutex && m_pTaskQueue && m_pIdleThreadStack && m_pActiveThreadList);for (int i = 0; i<nThreadAmount; i++) {CTaskThread *pMyThread = new CTaskThread(this);assert(pMyThread);PushIdleThread(pMyThread);pMyThread->StartThread();}return true;
}void CThreadPool::_Cleanup()
{if (m_pMyMutex) {delete m_pMyMutex;m_pMyMutex = NULL;}if (m_pTaskQueue) {m_pTaskQueue->Clear();delete m_pTaskQueue;m_pTaskQueue = NULL;}if (m_pIdleThreadStack) {m_pIdleThreadStack->Clear();delete m_pIdleThreadStack;m_pIdleThreadStack = NULL;}if (m_pActiveThreadList) {m_pActiveThreadList->Clear();delete m_pActiveThreadList;m_pActiveThreadList = NULL;}
}bool CThreadPool::_ReallocateThreads()
{m_pMyMutex->Lock();int nActiveThread = m_nThreadTotal - m_pIdleThreadStack->Size();if (m_nThreadActiveLow >= nActiveThread && m_nThreadActiveHigh >= m_nThreadTotal) {// If the number of active threads is insufficient and the total number of threads don't exceed the limit, the threads will be allocated again_Initialize(m_nCpuCoreTotal);m_nThreadTotal += m_nCpuCoreTotal;}m_pMyMutex->Unlock();return true;
}bool CThreadPool::SetCallbackFun(void *pCallback)
{if (pCallback) {m_fDecodeResult = (DecodeResult)pCallback;return true;}else {return false;}
}void CThreadPool::SetDecodeResult(DWORD dwResult)
{char szResult[10] = { 0 };_ltoa_s(dwResult, szResult, 10);m_fDecodeResult(szResult, strlen(szResult));
}

三、测试

3.1 demo源码

main.cpp

#include <stdio.h>// add necessary includes here
#include "../ThreadPool/ImageProcess.h"
#include <iostream>
using   namespace   std;#ifdef _DEBUG
#pragma comment(lib, "../Debug/ThreadPool.lib")
#else
#pragma comment(lib, "../Release/ThreadPool.lib")
#endifvoid myDecodeResult(char *pResultBuf, unsigned int unResultBufSize)
{printf("myDecodeResult() barcode=%s\n", pResultBuf);
}int main(void)
{char szPath[260] = "E:\\images\\";FeedbackResult(myDecodeResult);for (int i = 0; i < 20; i++) {PushImages(szPath, sizeof(szPath));_sleep(100);}//printf("fnPushImage() %d %s\n", nResult, szPath);getchar();return 0;
}

3.2 demo演示结果

New a thread: 21100
New a thread: 28040
New a thread: 11524
New a thread: 16328
New a thread: 10252
New a thread: 26128
New a thread: 24076
New a thread: 22124
New a thread: 25088
New a thread: 7236
Current thread: 7236, Current task: 14971584
New a thread: 17012
New a thread: 17208
New a thread: 9744
New a thread: 27136
Current thread: 27136, Current task: 14970864
Current thread: 9744, Current task: 14971056
Current thread: 17208, Current task: 14970912
Current thread: 17012, Current task: 14970960
Current task: 14971584, Rand result: 542441015
myDecodeResult() barcode=542441015
Current thread: 25088, Current task: 14971200
Current task: 14970864, Rand result: 542441140
myDecodeResult() barcode=542441140
Current thread: 22124, Current task: 14971584
Current task: 14971056, Rand result: 542441250
myDecodeResult() barcode=542441250
Current thread: 24076, Current task: 14971008
Current task: 14970912, Rand result: 542441359
myDecodeResult() barcode=542441359
Current thread: 26128, Current task: 14970864
Current task: 14970960, Rand result: 542441453
myDecodeResult() barcode=542441453
Current thread: 10252, Current task: 14970960
Current task: 14971200, Rand result: 542441562
myDecodeResult() barcode=542441562
Current thread: 16328, Current task: 14971392
Current task: 14971584, Rand result: 542441671
myDecodeResult() barcode=542441671
Current thread: 11524, Current task: 14971296
Current task: 14971008, Rand result: 542441781
myDecodeResult() barcode=542441781
Current thread: 28040, Current task: 14971584
Current task: 14970864, Rand result: 542441890
myDecodeResult() barcode=542441890
Current thread: 21100, Current task: 14971104
Current task: 14970960, Rand result: 542442000
myDecodeResult() barcode=542442000
Current task: 14971392, Rand result: 542442109
myDecodeResult() barcode=542442109
Current task: 14971296, Rand result: 542442218
myDecodeResult() barcode=542442218
Current task: 14971584, Rand result: 542442328
myDecodeResult() barcode=542442328
Current task: 14971104, Rand result: 542442437
myDecodeResult() barcode=542442437

总结

线程池使用可以在一定程序提高程序处理性能，如果使用不当，可能会造成死锁，且很难定位到错误点，应当谨慎使用。

C++线程池的创建与应用相关推荐

谈谈java的线程池（创建、机制）
目录 Executors创建线程池默认方法自定义线程池 Executors创建线程池默认方法 newFixedThreadPool()方法,该方法返回一个固定数量的线程池,该方法的线程数始终不变,当 ...
java(线程池的创建方式,和线程池的原理)
1.为什么要使用线程池: 减少资源消耗,通过重复的使用已创建好的线程,避免了线程的频繁创建和销毁所造成的消耗提高响应速度,当任务到达的时候,不需要再去创建,可以直接使用已经创建好的线程就能立即执 ...
Java同步—线程池的创建和使用
线程池构建一个新的线程是有一定代价的,因为涉及到与操作系统的交互.如果程序中需要使用大量生命周期很短的线程,就应该使用线程池. 将Runnable对象交给线程池来执行,就会有一个线程调用run方法, ...
线程池的创建及参数设置详解
一. 常见线程池线程池的创建方法主要有两类,第一是通过Executors 创建线程池,第二是通过 ThreadPoolExecutor 创建线程池. 首先我们来看通过Executors 创建的线程池 ...
JAVA线程池ThreadPoolExecutor创建以及各参数的详细说明
最近把线程很多的东西都温故了一遍,发现还漏了个线程池,今天看了些线程池的文章,然后加入了自己的理解,总结如下首先看下一个线程池的最简单的构造方法如下 * @param corePoolSize th ...
Linux下通用线程池的创建与使用
Linux下通用线程池的创建与使用本文给出了一个通用的线程池框架,该框架将与线程执行相关的任务进行了高层次的抽象,使之与具体的执行任务无关.另外该线程池具有动态伸缩性,它能根据执行任务的轻重自动调整 ...
SpringBoot线程池的创建、@Async配置步骤及注意事项
点击上方"方志朋",选择"设为星标" 回复"666"获取新整理的面试资料作者:Muscleheng blog.csdn.net/Muscl ...
Spring Boot 线程池的创建、@Async 配置步骤及注意事项
点击上方好好学java ,选择星标公众号重磅资讯.干货,第一时间送达今日推荐:终于放弃了单调的swagger-ui了,选择了这款神器-knife4j个人原创+1博客:点击前往,查看更多作者 ...
JUC多线程：线程池的创建及工作原理和 Executor 框架
一.什么是线程池: 线程池主要是为了解决新任务执行时,应用程序为任务创建一个新线程以及任务执行完毕时,销毁线程所带来的开销.通过线程池,可以在项目初始化时就创建一个线程集合,然后在需要执行新任务 ...
java线程池怎么创建_java中的线程池，如何创建？
Java中的线程池它是线程的容器,或者(换句话说,它是具有执行任务能力的线程的集合). 我们可以使用ThreadPool框架来定位(或实现)线程池. 线程池可以包含多个线程.每当我们执行任何任务时,线 ...

C++线程池的创建与应用

前言

一、线程池的设计原理

1.1 概念

1.2 优缺点

二、线程池的实现

2.1 业务流程图

2.2 类的介绍

2.3 类间关系图

2.4 类的实现

2.4.1 IThreadPool类

2.4.2 ITask类

2.4.3 CMyMutex类

2.4.4 CMyThreadList类

2.4.5 CMyThreadStack类

2.4.6 CMyTaskQueue类

2.4.7 CTaskThread类

2.4.8 CThreadPool类

三、测试

3.1 demo源码

3.2 demo演示结果

总结

C++线程池的创建与应用相关推荐

最新文章

热门文章