服务器设计系列：线程

废话不多说，详细介绍使用线程的优点好处请参考baidu、google。

一、线程使用场景

使用线程的方式大致有两种：

（1）流水线方式。根据业务特点，将一个流程的处理分割成多个线程，形成流水线的处理方式。产生的结果:延长单一流程的处理时间，提高系统整体的吞吐能力。

（2）线程池方式。针对处理时间比较长且没有内蕴状态的线程，使用线程池方式分流消息，加快对线程消息的处理，避免其成为系统瓶颈。

线程使用的关键是线程消息队列、线程锁、智能指针的使用。其中以线程消息队列最为重要。所谓线程消息队列，就是一个普通的循环队列（其它数据结构也未尝不可）加上多生产者－单（多）消费者的PV操作。流水线方式中的线程是单消费者，线程池方式中的线程是多消费者。

假定循环队列CircleQueue中，存放的消息指针类型是MyMSG *，入队操作EnQueue，出队操作DeQueue，判断队满IsQueueFull，判断队空IsQueueEmpty。生产者和消费者的流程通常是这样的：生产者线程生产消息(MyMSG *)，放在一个空缓冲区中(CircleQueue)，供消费者线程消费，生产者生产消息(EnQueue)，如果缓冲区满(IsQueueFull)，则被阻塞，消费者消费消息(DeQueue)，如果缓冲区空(IsQueueEmpty)，则被阻塞。线程消息队列就是生产者消费者问题中的缓冲区，而它的生产者是不限定的，任何线程都可以作为生产者向其中进行EnQueue操作，消费线程则可能是一个，也可能是多个。因此对循环队列的任何操作都要加锁，以保证线程安全。

二、Linux内核的线程库

2.6内核的默认安装的是redhat公司的NPTL（原生posix线程库），以前内核安装的是LinuxThreads库，两者的简单介绍可以看http://www.ibm.com/developerworks/cn/linux/l-threading.html。不过对于应用者，分析两者的区别和优劣也没什么大意义。这里特别提下NPTL的futex机制。借助该机制，pthread_mutex的性能大大提高，只要不进入竞争态，进程就不会陷入内核态。这点可以自己写示例程序，通过strace -c 跟踪进程的系统调用得以证实，另外还可以证实总是进入内核态的操作有pthread_cond_signal和sem_post。

线程的相关操作包括以下几种：

（1）线程类型pthread_t的创建、属性创建设置等。包括pthread_creat; pthread_attr_init; pthread_detach; pthread_join等，用法查看man。
（2）线程锁类型pthread_mutex_t的操作。包括pthread_mutex_init，查看man可看到所有相关的操作。
（3）条件变量类型pthread_cond_t的操作。包括pthread_cond_init。pthread_cond_t的wait和signal操作一定要和pthread_mutex_t的lock、unlock配合使用。类似于此：

pthread_mutex_t mux=PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond=PTHREAD_COND_INITIALIZER;

//wait操作：
pthread_mutex_lock(&mux);
pthread_cond_wait(&cond,&mux);//睡眠前将内部会执行pthread_mutex_unlock，醒来时内部会执行pthread_mutex_lock
pthread_mutex_unlock(&mux);

//signal操作
pthread_mutex_lock(&mux);
pthread_cond_signal(&cond);
pthread_mutex_unlock(&mux);

（4）信号号类型sem_t的操作。PV操作和锁机制的基础都是信号量。下面列出posix标准中给出的有关信号量的操作：

#include <semaphore.h>
int sem_init(sem_t *sem, int pshared, unsigned int value);
int sem_wait(sem_t * sem);
int sem_trywait(sem_t * sem);
int sem_post(sem_t * sem);
int sem_getvalue(sem_t * sem, int * sval);
int sem_destroy(sem_t * sem);

各个函数的详细用法，请在linux/unix上查看man。信号量通常用的不多，它比较重量级，但功能也最强大。

三、线程消息队列实现
通过将线程和线程消息队列封装在一起，可形成带有消息队列的线程。其它线程向该线程的消息队列插入消息，本线程取消息处理，之后再向其它线程的消息队列插入消息，如此形成流水线运行方式。下面设计一个简单的具有线程消息队列的线程封装类。通过上面的分析，我们可以有如下结论：
* 减少pthread_cond_signal和sem_post的调用，只在有必要的时候调用；
* 尽量避免pthread_mutex进入竞争态。增大消息队列的大小，可以有效减少竞态条件的出现。

代码如下，实现时使用了线程锁：

class CThreadQueue
{
private:
pthread_mutex_t mux;
pthread_cond_t condGet;
pthread_cond_t condPut;

void **buffer;    //循环消息队列
int sizeQueue;      //队列大小
int lput;           //location put 放数据的指针偏移
int lget;           //location get 取数据的指针偏移
int nFullThread;    //队列满，阻塞在putq处的线程数
int nEmptyThread;   //队列空，阻塞在getq处的线程数
int nData;          //队列中的消息个数，主要用来判断队列空还是满
public:
CThreadQueue(int queueSize=1024):
sizeQueue(queueSize),lput(0),lget(0),nFullThread(0),nEmptyThread(0),nData(0)
{
pthread_mutex_init(&mux,0);
pthread_cond_init(&condGet,0);
pthread_cond_init(&condPut,0);
buffer=new (void *)[sizeQueue];
}
virtual ~CThreadQueue()
{
delete[] buffer;
}
void * getq()  //出队操作
{
void *data;
pthread_mutex_lock(&mux); //加锁
//此处循环判断的原因如下：假设2个线程在getq阻塞，然后两者都被激活，而其中一个线程运行比较块，快速消耗了2个数据，
//另一个线程醒来的时候已经没有新数据可以消耗了。注意pthread_cond_wait是信号安全的系统调用，不会被信号中断。
while(lget==lput && nData==0)
{
nEmptyThread++;
pthread_cond_wait(&condGet,&mux);
nEmptyThread--;
}

data=buffer[lget++];
nData--;
if(lget==sizeQueue)
{
lget=0;
}
if(nFullThread) //必要时才进行signal操作，勿总是signal
{
pthread_cond_signal(&condPut);
}
pthread_mutex_unlock(&mux); //解锁
return data;
}
void putq(void *data)  //入队操作
{
pthread_mutex_lock(&mux);
while(lput==lget && nData)
{
nFullThread++;
pthread_cond_wait(&condPut,&mux);
nFullThread--;
}
buffer[lput++]=data;
nData++;
if(lput==sizeQueue)
{
lput=0;
}
if(nEmptyThread)
{
pthread_cond_signal(&condGet);
}
pthread_mutex_unlock(&mux);
}
};

一般系统设计中应该尽量减少锁的使用。但有的时候无法避免，这时就是mutex登场的时候了。mutex的实现，Linux下有pthread_mutex_t，ACE里有ACE_Thread_Mutex，boost里有boost::mutex。为了高效的操作可以进一步实现出其它不同的锁机制，比如常见的读写锁，条件锁，在Linux/ACE/boost中这些锁均有实现。lock和unlock要成对使用，但是很多情况下，一个函数有很多出口，再加上异常的情况，需要针对一个lock写很多unlock，这样不仅容易遗漏unlock，而且代码也变得很丑陋。ACE中提供了Guard封装mutex，使用起来比较方便，使用的时候不需要关心锁的释放，具体请看ACE。

在使用线程消息队列时，线程的创建可以使用posix的pthread_create函数，或者boost的boost::thread。具体使用请查看相关文档。另ACE中的ACE_Task实现了带有消息队列的线程，可以直接使用。下面给出这个线程消息队列的一个使用举例：

#include <pthread.h>
#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
CThreadQueue queue;//使用的时候给出稍大的CThreadQueue初始化参数，可以减少进入内核态的操作
void * produce(void * arg)
{
int i=0;
pthread_detach(pthread_self());
while(i++<100)
{
queue.putq((void *)i);
}
}
void *consume(void *arg)
{
int data;
while(1)
{
data=(int)(queue.getq());
printf("data=%d\n",data)
}
}
int main()
{    pthread_t pid;
int i=0;

while(i++<3)
pthread_create(&pid,0,produce,0);
i=0;
while(i++<3)
pthread_create(&pid,0,consume,0);
sleep(300);
}

四、使用智能指针的需求

在线程池方式中，为了去掉内蕴状态，线程间不得不传递对象指针，这样很难判断指针的生命周期，难以找到释放内存空间的合适位置。智能指针完美解决了这个问题。boost中有boost::shared_ptr,ACE中有ACE_Refcounted_Auto_Ptr。我们主要讲述线程相关，智能指针就不再展开了。
五、线程间消息传递框架
（1）面向过程的消息传递。C语言常用方式。消息以结构体的形式定义。

enum MsgType
{
CONCRETE_MSG1=1,
CONCRETE_MSG2=2
};
struct MyMsg
{
MsgType type;
union union_st
{
concreteMsg1 *msg1;
concreteMsg2 *msg2;
}msg;
};

concreteMsg1和concreteMsg2的详细结构不再列出。消息发送线程构建正确的具体消息，指明正确的消息类型，进一步构建正确的MyMsg，发送到线程消息队列。消息处理线程在消息队列头端循环getq，取出消息，根据消息类型调用相应的方法处理。

（2）面向对象的消息传递。线程消息队列中存储command模式中ICommand类型的指针。消息发送线程实例化具体的command，消息处理线程取出command执行command的execute方法。
缺点是command比较多的时候，会生成大量的类文件，代码不够紧凑。优点则是可以方便的增加command而不需要过多改动已有代码。

参考文献：

【原创】技术系列之 线程（二）：http://www.cppblog.com/CppExplore/archive/2008/03/20/44949.html