首页 > 代码库 > Loki之ThreadPool
Loki之ThreadPool
Loki中的ThreadPool目的主要是对创建出来的线程进行复用。
ThreadPool在Test而非Loki目录下，因此并非是标准Loki的组件之一，不过我们可以对其修改定制，
下面是对其源码的大致分析，ThreadPool顾名思义线程池，一般我们使用线程的时候CreateThread调用我们的回调函数，当回调函数结束之后我们的线程也随之终结销毁，
这样出现的问题是我们想执行一个多线程任务都要CreateThread，我们是否能够CreateThread调用我们的回调函数结束之后暂时将这个线程保存
起来而不是直接调回，当下次我们想要执行一个多线程任务的时候只需要把回调函数传递给该多线程对象，让它再次调用我们新的回调函数，这个
就类似与STL中的内存池对内存的复用，只是STL内存池会对整块大内存进行分片链接到链表处理而已。
想要真正了解ThreadPool最好的办法是浏览源代码
首先是线程对象Thread，它有以下几种状态：
enum Status
{
Dead = 0,
Idle,
Starting,
Active
};
在创建线程并启动的时候起内部调用回调函数是
#if defined( _MSC_VER )
unsigned int ThreadPool::TopFunction( void * p )
#else
void * ThreadPool::TopFunction( void * p )
#endif
{
assert( nullptr != p );
volatile Thread * thread = reinterpret_cast< volatile Thread * >( p );
Thread::SetCurrentThread( thread );
while ( ( thread->m_status != Thread::Dead ) && ( !thread->m_stop ) )
{
// Call the thread‘s WaitPolicy here?
#if defined( _MSC_VER )
::SleepEx( 1, true );
#else
::sleep( 1 );
#endif
if ( thread->m_status == Thread::Starting ) //如果线程状态变为Starting，说明要复用该线程，调用用户的回调函数
{
try
{
assert( nullptr != thread->m_func );
thread->m_status = Thread::Active;
thread->m_func( thread->m_parm );
}
catch ( ... )
{
// What to do in case of exception?
// Call an exception policy?
}
//用户回调函数运行结束后将线程状态设置为空闲，以便下次复用
thread->m_status = Thread::Idle;
thread->m_func = nullptr;
thread->m_parm = nullptr;
}
}
#if defined( _MSC_VER )
return 0;
#else
return nullptr;
#endif
}
//该函数判读的是用户的回调函数是否执行结束，而非ThreadPool传递给它的TopFunction是否结束，
//返回true仅仅说明用户的回调函数执行结束，实际该线程依然存在于内存池中
bool Thread::WaitForThread( void ) volatile
{
assert( IsValid( m_owner ) );
const volatile Thread * current = Thread::GetCurrentThread();
if ( this == current )
return false;
if ( m_status == Thread::Dead )
return false;
while ( this->m_status == Thread::Active )
{
// Call the wait policy.
#if defined( _MSC_VER )
::SleepEx( 1, true );
#else
::sleep( 1 );
#endif
}
return true;
}
下面介绍ThreadPool，ThreadPool使用vector对线程进行保存，主要列举关键的几个函数进行说明
//创建threadCount个多线程于内存池中
unsigned int ThreadPool::Create( unsigned int threadCount ) volatile
{
assert( IsValid() );
LOKI_DEBUG_CODE( Checker checker( this ); (void)checker; )
ThreadPool * pThis = const_cast< ThreadPool * >( this );
//计算内存池中Idle状态的线程数目，如果不够客户要求则创建足够的线程放到线程池中
const unsigned int countNow = GetCount( Thread::Idle );
if ( threadCount <= countNow )
return threadCount;
const unsigned int totalCount = pThis->m_threads.size();
const unsigned int howManyToAdd = threadCount - countNow;
if ( pThis->m_threads.capacity() <= howManyToAdd )
pThis->m_threads.reserve( totalCount + howManyToAdd );
for ( unsigned int ii = 0; ii < howManyToAdd; ++ii )
{
#if defined( _MSC_VER )
volatile Thread * thread = new Thread( this );
#else
Thread * thread = new Thread( this );
#endif
pThis->m_threads.push_back( thread );
Thread * pThread = const_cast< Thread * >( thread );
void * p = reinterpret_cast< void * >( pThread );
// Call thread creation policy? 此处可以调用创建线程策略，留给读者进行改进
LokiThreadCreate( &thread->m_thread, nullptr, TopFunction, p );
}
return howManyToAdd;
}
//开始执行用户的回调函数，function为用户的回调函数，parm为传递给function的void指针
volatile Thread * ThreadPool::Start( CallFunction function, void * parm ) volatile
{
assert( IsValid() );
LOKI_DEBUG_CODE( Checker checker( this ); (void)checker; )
ThreadPool * pThis = const_cast< ThreadPool * >( this );
if ( nullptr == function )
return nullptr;
#if defined( _MSC_VER )
volatile Thread * thread = nullptr;
#else
Thread * thread = nullptr;
#endif
bool foundOne = false;
//查找内存池中是否有idle线程
for ( size_t ii = 0; ii < pThis->m_threads.size(); ii++ )
{
#if defined( _MSC_VER )
thread = pThis->m_threads.at( ii );
#else
thread = const_cast< Thread * >( pThis->m_threads.at( ii ) );
#endif
assert( nullptr != thread );
if ( Thread::Idle == thread->m_status )
{
foundOne = true;
break;
}
}
//存在空闲线程的话进行复用，否则创建一个线程并放入内存池vector中，
//对相应参数赋值，设置线程状态为Starting之后在内存池回调函数TopFunction中会自动调用用户回调函数function，线程状态设置为active，
//function结束之后在将线程状态设置为idle，这样就可以达到线程复用的效果
if ( foundOne )
{
thread->m_func = function;
thread->m_parm = parm;
thread->m_status = Thread::Starting;
}
else
{
// Did not find an idle thread, so start a new one.
thread = new Thread( this, function, parm );
pThis->m_threads.push_back( thread );
Thread * pThread = const_cast< Thread * >( thread );
void * p = reinterpret_cast< void * >( pThread );
// Call to thread creation policy?
LokiThreadCreate( &thread->m_thread, nullptr, TopFunction, p );
}
return thread;
}
该线程池比较容易理解与使用，但是觉得不太好的地方就是TopFunction中循环的SleepEx部分，以及线程创建之后必须要再ThreadPool析构之后才会销毁，在Start的时候如果找不到idle线程
就会创建一个新的线程push_back到vector中，不能够保证控制线程在一个数量下。
设计改进：
用户希望内存池最多用n个线程，Start函数根据当前的线程数量可以在找不到idle线程的时候创建线程，但是达到n之后不能够在创建，这样我们可以把用户的function和parm保存
到一个队列当中，当线程池中的线程有idle线程的时候我们从任务队列中取出元素执行之
Loki之ThreadPool