线程池实例讲解（原创）

　　最近老看一些服务器网关的代码页看了一些开源的代码。一个重要的技术线程池。

　　何为线程池，所谓线程池就是一组用来处理,客户请求的线程组 这里的客户指代的是线程池服务的对象。

　　

　　线程池的实现原理：

　　　　（1）消息队列调用函数，当有消息到来时候，将消息封装插入消息队列。

　　　　（2）有一个 线程池附服务线程，该线程负责检索消息队列，创建线程池线程，将该消息派发到线程池的某一线程处理。

　　　　（3）线程池线程参数，该参数应该是一个结构体，该结构体中某项指向一个消息，处理完成之后删除消息，进入等待状态。

　 

　　源代码如下 代码注释有说明 故不在单独讲解代码：

#include <stdlib.h>
#include <stdio.h>
#include <unistd.h>
#include <assert.h>
#include <fcntl.h>
#include <string.h>
#include <time.h>
#include <pthread.h>
#include <sys/types.h>


//消息队列基本元素
typedef  struct _STPOOLMsgElement       //消息队列基本元素
{
   char * param;                       //代表消息队列消息  实际情况中也可以使用结构体
   struct _STPOOLMsgData *next;         //下一条消息
   
}STPOOLMsgData,*PSTPOOLMsgData;


//线程池基本元素
typedef struct _STPOOLThreadElement
{
    pthread_t thread;                      //但前线程ID
    pthread_mutex_t busy;                  // 用来同步线程池线程 和线程池服务线程
    pthread_cond_t nodata;                 //
    PSTPOOLMsgData pMsgData;               //当前线程需要处理的事件，如果为空线程池空闲
    struct _STPOOLThreadElement *next;     //线程池的下一个线程

}STPOOLThreadElement,*PSTPOOLThreadElement;

//变量定义
//消息队列肯定有头有尾头出尾巴进  多个线程要访问必须有同步 以及初始化
PSTPOOLMsgData     pStMsgQuequeBegin = NULL;
PSTPOOLMsgData     pStMsgQuequeEnd = NULL;
pthread_mutex_t queueAccess ;//= PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t     queueEmpty; //= PTHREAD_COND_INITIALIZER;

//线程池链表  线程池的线程不管先后顺序，每次任务来了检索空闲线程，派遣任务
PSTPOOLThreadElement  pStPool = NULL;

//先来看消息队列相关函数  消息插入函数 负责消息封装插入消息队列
void InsertMsg(char * pMsg)
{
    //将参数封装成消息队列元素
    PSTPOOLMsgData pMsgData = (PSTPOOLMsgData)malloc(sizeof(STPOOLMsgData));
    if(!pMsgData)  return;
    
    char *pMsgDupli = malloc(strlen(pMsg) + 1);   //消息数据
    if(!pMsgDupli)
    {
        free(pMsgData);
        return;
    }
    strcpy(pMsgDupli,pMsg);
    pMsgData->param = pMsgDupli;
    pMsgData->next = NULL;

    //插入消息
    pthread_mutex_lock(&queueAccess);
    if (pStMsgQuequeEnd)
    {
        pStMsgQuequeEnd->next = pMsgData;
    }
    pStMsgQuequeEnd = pMsgData;

    if (!pStMsgQuequeBegin)
    {
        pStMsgQuequeBegin = pMsgData;
    }

    pthread_cond_broadcast(&queueEmpty);    //通知消息分发线程有消息到来
    pthread_mutex_unlock(&queueAccess);
}


//再来看看消息取出函数  采用阻塞模式，因为没有消息的时候所有线程都处于空闲状态
PSTPOOLMsgData GetNextMsg()
{
    PSTPOOLMsgData Result = NULL;

    pthread_mutex_lock(&queueAccess);
    //如果没有事件，一直等待直道有事件产生    取出一个事件
    while (pStMsgQuequeBegin == NULL)
        pthread_cond_wait(&queueEmpty, &queueAccess);

    Result = pStMsgQuequeBegin;
    pStMsgQuequeBegin = Result->next;
    if (!pStMsgQuequeBegin)     //没有消息
    {
        pStMsgQuequeEnd = NULL;
    }
    Result->next = NULL;
    pthread_mutex_unlock(&queueAccess);
    return Result;
}


//接下来看线程池线程
void *PoolLDealMsgThread(PSTPOOLThreadElement listElement)
{
    int threadId = 0;
    pthread_mutex_lock(&listElement->busy);
    while (1)
    {
        while (listElement->pMsgData == NULL)    //无事件处理 一直阻塞等待
            pthread_cond_wait(&listElement->nodata, &listElement->busy);

        //处理事件，清除事件占用空间
        
        threadId = listElement->thread;
        printf("threadID :%d  Msg: %s 
",threadId,listElement->pMsgData->param);
        
        free(listElement->pMsgData->param);
        free(listElement->pMsgData);

        listElement->pMsgData = NULL;
    }
    pthread_mutex_unlock(&listElement->busy);
    pthread_exit(NULL);
    return NULL;
}


//接着看看线程池服务线程 这里调用DispatchMsg 函数分发消息
void DispatchMsg(PSTPOOLMsgData pMsgData)
{
       int Result = 0;
       PSTPOOLThreadElement cur = NULL;

       //查找空闲线程处理消息
       for (cur = pStPool; cur != NULL; cur = cur->next)
       {
            //双重锁定   提高程序性能
            if (cur->pMsgData == NULL)
            {
                pthread_mutex_lock(&cur->busy);
                if (cur->pMsgData != NULL)
                {
                    pthread_mutex_unlock(&cur->busy);
                    continue;
                }
                cur->pMsgData = pMsgData;
                pthread_cond_broadcast(&cur->nodata);   //通知线程池线程有事件到来
                pthread_mutex_unlock(&cur->busy);
                return;
            }
      }
      //新建线程处理消息
      cur = (PSTPOOLThreadElement)malloc(sizeof(STPOOLThreadElement));
      if(!cur)
      {
            free(pMsgData->param);
            free(pMsgData);
            return;
      }

      memset(cur,0,sizeof(STPOOLThreadElement));

      pthread_mutex_init(&cur->busy, NULL);
      pthread_cond_init(&cur->nodata, NULL);
      cur->pMsgData = pMsgData;
      cur->next = NULL;
      Result = pthread_create(&cur->thread, NULL, (void *(*)(void *)) PoolLDealMsgThread, cur);

      if (0 != Result)
      {
        free(cur);
        free(pMsgData->param);
        free(pMsgData);
        return;
      }
      //新线程插入线程池
      cur->next = pStPool;
      pStPool = cur;
}


void *QueueDealFuction(void *data)
{
    PSTPOOLMsgData pMsgData = NULL;
    while (1)
    {
        pMsgData = GetNextMsg();  //读取一条事件记录
        DispatchMsg(pMsgData);
    }
    return NULL;
}

//好了基本线程池模型已经构建完毕 现在看看主函数
char testMsg[5][10]  = {"hello!",
                    "this",
                    "is",
                    "my",
                    "testApp",
                    } ;

int main()
{
    //初始化开启线程池服务线程
    int i = 0;
    pthread_mutex_init(&queueAccess, NULL);
    pthread_cond_init(&queueEmpty, NULL);
    
    pthread_t queueHandler;
    if(0 != pthread_create(&queueHandler, NULL, (void *(*)(void *)) QueueDealFuction, NULL) )
    {
        return -1;
    }
    //这里消息仅代表字符串 
    for(i=0 ;i < 5 ;i++)
    {
        InsertMsg(testMsg[i]);
    }
    sleep(10);
    return 0;
}

编译

gcc -g  main.c -lpthread  -o main

连续运行程序三次 三次结果如下：

 

分析结果可以知道： 在线程池中并不是先分配处理线程的的消息先处理完成，这取决于操作系统线程调度。

由此可以得知如要顺序处理消息，这种方式不使用于线程池。

一些技巧：

　　线程池常用于服务器网络编程，线程池参数中往往携带输出socket ，当我们处理完消息后，将结果通过socket 发送给客户端。

　　如：web服务器，加载生成XML文件，通过socket 发送给客户端。