信号量
#include<semaphore.h>
int sem_init(sem_t* sem, int pshared, unsigned int value);
int sem_destory(sem_t* sem);
int sem_wait(sem_t* sem);
int sem_trywait(sem_t* sem);
int sem_post(sem_t* sem);
这些函数的第一个参数sem指向被操作的信号量。
- sem_init函数用于初始化一个未命名的信号量。pshared参数指定信号量的类型。如果其值为0，就表示这个信号量是当前进程的局部信号量，否则该信号量就可以在多个进程之间共享。value参数指定信号量的初始值。此外，初始化一个已经被初始化的信号量，将会导致不可预期得结果。
- sem_destory函数用于销毁信号量，以释放其占用的内核资源。如果销毁一个正在被其他线程等待的信号量，将会导致不可预期的结果。
- sem_wait函数以原子操作的方式将信号量的值减1.如果信号量的值为0，则sem_wait将被阻塞，直到这个信号量具有非0值。
- sem_trywait函数与sem_wait函数相似，不过它始终立即返回，而不论被操作的信号量是否具有非0值，相当于sem_wait的非阻塞版本。当前信号量的值非0时，sem_trywait对信号量执行减1操作。当信号量的值为0时，它将返回-1并设置errno为EAGAIN。
- sem_post函数以原子操作的方式将信号量的值加1。当信号量的值大于0时，其他正在调用sem_wait等待信号量的线程将被唤醒。
上面这些函数成功时返回0，失败则返回-1，并设置errno。
一个初始值为Ｎ的信号量允许Ｎ个线程并发访问。线程访问资源的时候首先获取信号量，进行如下操作
- 将信号量的值减-1 (sem_wait函数)
- 如果信号量的值小于０，则进入等待状态，否则继续执行。
当访问完资源后，线程释放信号量，进行如下操作。 - 将信号量的值加１(sem_post函数)．
- 如果信号量的值小于１，唤醒一个等待的线程。
互斥锁
互斥锁也称互斥量，可以用于保护关键代码段，以确保其独占式的访问。当进入关键代码段时，我们需要获得互斥锁并将其加锁；当离开代码段时，我们需要对互斥锁解锁，以唤醒其他等待该互斥量的线程。
互斥锁基础API
#include<pthread.h>
int phtread_mutex_init(pthread_mutex_t* mutex, const pthread_mutexattr_t* mutexattr);
int pthread_mutex_destroy(pthread_mutex_t* mutex);
int phthread_mutex_lock(pthread_mutex_t* mutex);
int phtread_mutex_trylock(phtread_mutex_t* mutex);
int phtread_mutex_unlock(phtread_mutex* mutex);
这些函数的第一个参数mutex指向要操作的目标互斥锁，互斥锁的类型时pthread_mutex_t结构体。
- pthread_mutex_init函数用于初始化互斥锁。mutexattr参数指定互斥锁的属性。如果将它设置为NULL,则表示使用默认属性。我们还可以用如下方式来初始化一个互斥锁：
pthread_mutex_t mutex = PHTREAD_MUTEX_INITIALIZER;
//宏PHTREAD_MUTEX_INITIALIZER把互斥锁的各字段都初始化为0.
- pthread_mutex_lock函数以原子操作的方式给一个互斥锁加锁。如果目标互斥锁已经被加锁，则该函数调用将阻塞住，直到该互斥锁的占有者将其解锁。
- pthread_mutex_trylock与上条函数类似，不过它始终立即返回，而不论被操作的互斥锁是否已经被加锁，相当于pthread_mutex_lock函数的非阻塞版本。当目标互斥锁未被加锁时，则将其进行加锁。当互斥锁已经被加锁时，该函数返回错误码EBUSY。
- pthread_mutex_unlock函数以原子操作的方式给一个互斥锁解锁。如果此时有其他线程正在等待这个互斥锁，则这些线程中的某一个将获得它.
上面这些函数成功时返回0， 失败则返回错误码。
死锁情况举例
//死锁情况
#include<pthread.h>
#include<iostream>
#include<unistd.h>
int a = 0;
int b = 0;
pthread_mutex_t mutex_a;
pthread_mutex_t mutex_b;
void* anthor(void* arg)
{
pthread_mutex_lock(&mutex_b);
std::cout << "in child thread, got mutex b, waiting for mutex a\n";
sleep(5);
++b;
pthread_mutex_lock(&mutex_a);
b += a++;
pthread_mutex_unlock(&mutex_a);
pthread_mutex_unlock(&mutex_b);
pthread_exit(NULL);
}
int main()
{
pthread_t id;
//初始化锁
pthread_mutex_init(&mutex_a, NULL);
pthread_mutex_init(&mutex_b, NULL);
//创建线程
pthread_create(&id, NULL, anthor, NULL);
pthread_mutex_lock(&mutex_a);
std::cout << "in parent thread, got mutex a, waiting for mutex b\n";
sleep(5);
++a;
pthread_mutex_lock(&mutex_b);
a += b++;
pthread_mutex_unlock(&mutex_a);
pthread_mutex_unlock(&mutex_b);
//等待线程id结束
pthread_join(id, NULL);
//销毁锁
pthread_mutex_destroy(&mutex_a);
pthread_mutex_destroy(&mutex_b);
}
该程序运行结果，造成死锁，程序一直锁住无法继续执行。
我们来分析一下代码：
主线程试图先占有互斥锁mutex_a，然后操作该锁保护的变量a，但操作完毕之后，主线程并没有立即释放互斥锁mutex_a，而是又申请互斥锁mutex_b,并在两个互斥锁的保护下，操作变量a和b，最后才一起释放两个互斥锁；与此同时，子线程则按照相反的顺序来申请互斥锁mutex_a和mutex_b，并在两个锁的保护下操作变量a和b。我们用sleep函数来模拟连续两次调用pthread_mutex_lock之间的时间差，以确保代码中的两个线程各自先占有一个互斥锁（主线程占有互斥锁a，子线程占有互斥锁b），然后等待另一个互斥锁。这样，两个线程就僵持住了，谁也不能继续往下执行，从而形成死锁。如果代码不加入sleep函数，则这段代码有可能会成功运行，但是这种逻辑是一个典型的死锁错误。
条件变量
如果说互斥锁是用于同步线程对共享数据的访问的话，那么条件变量则是用于在线程之间同步共享数据的值。条件变量提供了一种线程间的通知机制；当某个共享数据达到某个值的时候，唤醒等待这个共享数据的线程。
条件变量相关函数主要有以下五个
#include<pthread.h>
int pthread_cond_init(pthread_cond_t* cond, const pthread_condattr_t* cond_attr);
int pthread_cond_destroy(pthread_cond_t* cond);
int pthread_cond_broadcast(pthread_cond_t* cond);
int pthread_cond_signal(pthread_cond_t* cond);
int phtread_cond_wait(pthread_cond_t* cond, phtread_mutex_t* mutex);
这些函数的第一个参数cond指向要操作的目标条件变量，条件变量的类型是pthread_cond_t结构体。
- pthread_cond_init函数用于初始化条件变量。cond_attr参数指定条件变量的属性。如果将它设置为NULL，则表示使用默认属性。
pthread_cond_t cond = PTHREAD_COND_INITIALIZER; //另一种初始化方式
- phtread_cond_destroy函数用于销毁条件变量，以释放其占有的内核资源。销毁一个正在被等待的条件变量将失败返回EBUSY。
- pthread_cond_broadcast函数以广播的方式唤醒所有等待目标条件变量的线程。
- pthread_cond_signal函数用于唤醒一个等待目标条件变量的线程。至于哪个线程将被唤醒，则取决于线程的优先级和调度策略。有时候我们可能想唤醒一个指定的线程，但pthread没有对该需求提供解决方法。不过我们可以间接地实现需求：定义一个能够唯一表示目标线程的全局变量，在唤醒等待条件变量的线程前先设置该变量为目标线程，然后采用广播方式唤醒所有等待条件变量的线程，这些线程被唤醒后都检查该变量以判断被唤醒的是否是自己，如果是就开始执行后续代码，如果不是则返回继续等待。
- pthread_cond_wait函数用于等待目标条件变量。mutex参数是用于保护条件变量的互斥锁，**以确保pthread_cond_wait操作的原子性。**在调用pthread_cond_wait前，必须确保互斥锁mutex已经加锁，否则将导致不可预期的结果。pthread_cond_wait函数执行时，**首先把调用线程放入条件变量的等待队列中，然后将互斥锁mutex解锁。**可见，从pthread_cond_wait开始执行到其调用线程被放入条件变量的等待队列之间的这段时间内，pthread_cond_signal和pthread_cond_broadcast等函数不会修改条件变量。换言之就是pthread_cond_wait函数不会错过目标条件变量的任何变化。当pthread_cond_wait函数成功返回时，互斥锁mutex将再次被锁上。
上面这些函数成功时返回0，失败则返回错误码
在生产者消费者模型中，如果我们的线程此时都去处理任务，这个时候有新的任务加入而没有线程在等待，此时生产者调用pthread_cond_signal函数会产生什么？
答：条件变量并不保存状态信息，只是传递应用程序状态信息的一种通讯机制。发送信号时，若无任何线程在等待该条件变量，这个信号也就会不了了之。线程如在此后等待该条件变量，只有当再次收到次变量的下一个信号时，方可解除阻塞状态。
我们将在下面代码中将以上三种线程同步机制封装为三个类。
//对信号量、互斥锁和条件变量封装为三个类
#ifndef LOCKER_H
#define LOCKER_H
#include<iostream>
#include<exception>
#include<pthread.h>
#include<semaphore.h>
//封装信号量的类
class sem
{
public:
//创建并初始化信号量
sem()
{
if(sem_init(&m_sem, 0, 0) != 0)
{
//构造函数没有返回值，可以通过抛出异常来报告错误
throw std::exception();
}
}
//销毁信号量
~sem()
{
sem_destroy(&m_sem);
}
//等待信号量
bool wait()
{
return sem_wait(&m_sem) == 0;
}
//增加信号量
bool post()
{
return sem_post(&m_sem) == 0;
}
private:
sem_t m_sem;
};
//封装互斥锁的类
class locker
{
public:
//创建并初始化互斥锁
locker()
{
if(pthread_mutex_init(&m_mutex, NULL) != 0)
throw std::exception();
}
//销毁互斥锁
~locker()
{
pthread_mutex_destroy(&m_mutex);
}
//获取互斥锁
bool lock()
{
return pthread_mutex_lock(&m_mutex) == 0;
}
//释放互斥锁
bool unlock()
{
return pthread_mutex_unlock(&m_mutex) == 0;
}
private:
pthread_mutex_t m_mutex;
};
//封装条件变量的类
class cond
{
public:
//创建并初始化条件变量
cond()
{
if(pthread_mutex_init(&m_mutex, NULL) != 0)
throw std::exception();
if(pthread_cond_init(&m_cond, NULL) != 0)
throw std::exception();
}
//销毁条件变量
~cond()
{
pthread_mutex_destroy(&m_mutex);
pthread_cond_destroy(&m_cond);
}
//等待条件变量
bool wait()
{
int ret = 0;
pthread_mutex_lock(&m_mutex);
ret = pthread_cond_wait(&m_cond, &m_mutex);
pthread_mutex_unlock(&m_mutex);
return ret == 0;
}
//唤醒等待条件变量的线程
bool signal()
{
return pthread_cond_signal(&m_cond) == 0;
}
private:
pthread_mutex_t m_mutex;
pthread_cond_t m_cond;
};
下面是我们用上面封装的锁和信号量实现的线程池结构
#ifndef THREADPOOL_H
#define THREADPOOL_H
#include<list>
#include<cstdio>
#include<exception>
#include<pthread.h>
#include"locker.h"
//线程池类,将它定义为模板类是为了代码复用。模板参数T是任务类
template<typename T>
class threadpool
{
public:
/*参数thread_number是线程池中线程的数量，max_requests是请求队列中最多允许的、等待处理请求的数量 */
threadpool(int thread_number = 8, int max_requests = 10000);
~threadpool();
//往请求队列中添加任务
bool append(T* request);
//工作线程运行函数
static void* worker(void* arg);
void run();
private:
int m_thread_number; //线程池中的线程数
int m_max_requests; //请求队列中允许的最大请求数
pthread_t* m_threads; //描述线程池的数组，其大小为m_thread_number
std::list<T*>m_workqueue; //请求队列
locker m_queuelocker; //保护请求队列的互斥锁
sem m_queuestat; //是否有任务需要处理
bool m_stop; //是否结束线程
};
template<typename T>
threadpool<T>::threadpool(int thread_number, int max_requests)
: m_thread_number(thread_number), m_max_requests(max_requests), m_stop(false), m_threads(NULL)
{
if((thread_number <= 0) || (max_requests <= 0) )
throw std::exception();
//创建线程池
m_threads = new pthread_t[m_thread_number];
if(!m_threads)
{
throw std::exception();
}
//创建thread_number个线程，并将它们设置为脱离线程
for(int i = 0; i < m_thread_number; ++i)
{
printf("create the %dth thread\n", i);
if(pthread_create(m_threads + i, NULL, worker, this) != 0)
{
delete [] m_threads;
throw std::exception();
}
if(pthread_detach(m_threads[i]))
{
delete [] m_threads;
throw std::exception();
}
}
}
template<typename T>
threadpool<T>::~threadpool()
{
delete [] m_threads;
m_stop = true;
}
//将任务添加到任务队列中
template<typename T>
bool threadpool<T>::append(T* request)
{
//操作工作队列是一定要加锁，因为它被所有线程共享
m_queuelocker.lock();
if(m_workqueue.size() > m_max_requests)
{
m_queuelocker.unlock();
return false;
}
m_workqueue.push_back(request);
m_queuelocker.unlock();
//将信号量加１，唤醒等待线程
m_queuestat.post();
return true;
}
template<typename T>
void* threadpool<T>::worker(void* arg)
{
threadpool* pool = (threadpool*)arg;
pool->run();
return pool;
}
template<typename T>
void threadpool<T>::run()
{
while(!m_stop)
{
//信号量减１，阻塞等待被唤醒
m_queuestat.wait();
m_queuelocker.lock();
//如果任务队列中无任务
if(m_workqueue.empty())
{
//解锁
m_queuelocker.unlock();
continue;
}
//将任务队列头取出
T* request = m_workqueue.front();
//将取出的任务从队列中删掉
m_workqueue.pop_front();
////
m_queuelocker.unlock();
if(!request)
{
continue;
}
request->process(); //执行回调函数，该函数未定义，根据大家的需求自己定义
}
}
#endif