Linux下线程经典问题(生产者消费者问题，哲学家问题...)

线程经典问题

本月任务
一. 解决生产者消费者问题
二. 解决哲学家吃饭问题
三. 实现进程池/线程池

在编译运行程序之后需要加上 -lpthread

生产者消费者问题

问题：生产者消费者共享缓冲区，生产者向缓冲区中放数据，消费者从缓冲取中取数据，当缓冲区中被放满时，生产者进程就必须进入挂起状态，直到消费者从缓冲中取走数据时，生产者才能继续向缓冲区中存放数据，同样当缓冲取中没有数据时，消费者进程就必须进入挂起休眠状态，直到生产者向缓冲区中放入数据时，消费者才能被唤醒继续从缓冲区中取走数据。

生产者消费者问题，也称有限缓冲问题，是一个多线程同步问题的经典案例。

解决这个问题之前，我们应该了解一下PV操作（这篇文章很好理解）

# include <stdio.h>
# include <pthread.h>
# include <unistd.h>
# include <stdlib.h>#define N 100
#define true 1
#define producerNum  10
#define consumerNum  5typedef int semaphore;
typedef int item;item buffer[N] = {0};
int in = 0;
int out = 0;
int proCount = 0;
semaphore mutex = 1, empty = N, full = 0, proCmutex = 1;void * producer(void * a)
{while(true) {while( proCmutex <= 0 );proCmutex--;proCount++;printf("生产一个产品ID%d, 缓冲区位置为%d\n",proCount,in);proCmutex++;while( empty <= 0 ) {printf("缓冲区已满！\n");}empty--;while( mutex <= 0 );mutex--;buffer[in] = proCount;in = (in + 1) % N;mutex++;full++;sleep(1);}
}void * consumer(void *b)
{while(true) {while( full <= 0 ) {printf("缓冲区为空！\n");}full--;while( mutex <= 0 );mutex--;int nextc = buffer[out];buffer[out] = 0;//消费完将缓冲区设置为0out = (out + 1) % N;mutex++;empty++;printf("\t\t\t\t消费一个产品ID%d,缓冲区位置为%d\n", nextc,out);sleep(1);}
}int main()
{pthread_t threadPool[producerNum+consumerNum];int i;for(i = 0; i < producerNum; i++) {pthread_t temp;if ( pthread_create(&temp, NULL, producer, NULL) == -1 ) {printf("ERROR, fail to create producer%d\n", i);exit(1);}threadPool[i] = temp;}//创建生产者进程放入线程池for(i = 0; i < consumerNum; i++) {pthread_t temp;if ( pthread_create(&temp, NULL, consumer, NULL) == -1 ) {printf("ERROR, fail to create consumer%d\n", i);exit(1);}threadPool[i+producerNum] = temp;}//创建消费者进程放入线程池void * result;for(i = 0; i < producerNum+consumerNum; i++) {if ( pthread_join(threadPool[i], &result) == -1 ) {printf("fail to recollect\n");exit(1);}}//运行线程池return 0;
}

哲学家吃饭问题

问题：有五个哲学家绕着圆桌坐，每个哲学家面前有一盘面，两人之间有一支筷子，这样每个哲学家左右各有一支筷子。哲学家有2个状态，思考或者拿起筷子吃饭。如果哲学家拿到一只筷子，不能吃饭，直到拿到2只才能吃饭，并且一次只能拿起身边的一支筷子。一旦拿起便不会放下筷子直到把饭吃完，此时才把这双筷子放回原处。如果，很不幸地，每个哲学家拿起他或她左边的筷子，那么就没有人可以吃到饭了。

哲学家进餐问题是一个多线程运用的经典例子，涉及到线程同步/互斥，临界区访问问题以及死锁问题。

吃饭时使用筷子：

拿一双筷子才能吃
每次只允许拿一支筷子
只能拿身边的筷子
吃完才放下筷子

5个哲学家可能每个人都拿起自己的左筷子，但是却无法拿到自己的右筷子。既无法释放自己的筷子，也等不到别人的筷子完成自己的活动，最终形成死锁。

死锁：两个或多个进程无限期地等待永远不会发生的条件的一种系统状态（结果：每个进程都永远阻塞）

在此问题中死锁为：每个哲学家都无限期的等待邻座哲学家放下筷子！
而邻座哲学家没有吃完饭前不会放下筷子！
邻座哲学家缺一支筷子永远无法吃完饭！

方法一：
每个哲学家对应一个线程，程序中定义一个互斥量，对于每个线程进行访问其他哲学家状态时用互斥量进行加锁，这样也就避免了死锁的产生，访问到该哲学家处于饥饿时，同时旁边两位科学家并未处于进餐状态时，他就拿起左右两边的叉子进行吃饭，吃饭一段时间后，就放下叉子进行思考，思考一段时间后处于饥饿状态，重新开始试图拿起叉子吃饭，代码如下

semaphore chopstick[5]={1，1，1，1，1};
semaphore room=4;
#include<sys/types.h>
#include<unistd.h>
#include<stdlib.h>
#include<stdio.h>
#include<pthread.h>
#include<semaphore.h>
#include<time.h>#define N 5     //哲学家数量#define LEFT(i)    (i+N-1)%N  //左手边哲学家编号
#define RIGHT(i)   (i+1)%N    //右手边哲家编号#define HUNGRY    0     //饥饿
#define THINKING  1     //思考
#define EATING    2     //吃饭#define U_SECOND 1000000   //1秒对应的微秒数
pthread_mutex_t mutex;     //互斥量int state[N];  //记录每个哲学家状态
//每个哲学家的思考时间，吃饭时间，思考开始时间，吃饭开始时间
clock_t thinking_time[N], eating_time[N], start_eating_time[N], start_thinking_time[N];
//线程函数
void *thread_function(void *arg);int main()
{pthread_mutex_init(&mutex, NULL);pthread_t a,b,c,d,e;//为每一个哲学家开启一个线程，传递哲学家编号pthread_create(&a,NULL,thread_function,"0");pthread_create(&b,NULL,thread_function,"1");pthread_create(&c,NULL,thread_function,"2");pthread_create(&d,NULL,thread_function,"3");pthread_create(&e,NULL,thread_function,"4");//初始化随机数种子srand((unsigned int)(time(NULL)));while(1){;}
}void *thread_function(void *arg)
{char *a = (char *)arg;int num = a[0] - '0';  //根据传递参数获取哲学家编号int rand_time; while(1) {//关键代码加锁pthread_mutex_lock(&mutex);//如果该哲学家处于饥饿  并且  左右两位哲学家都没有在吃饭  就拿起叉子吃饭if (state[num] == HUNGRY && state[LEFT(num)] != EATING && state[RIGHT(num)] != EATING) {state[num] = EATING;start_eating_time[num] = clock(); //记录开始吃饭时间eating_time[num] = (rand() % 5 + 5) * U_SECOND;   //随机生成吃饭时间//输出状态printf("state: %d %d %d %d %d\n",state[0],state[1],state[2],state[3],state[4]);//printf("%d is eating\n",num);} else if (state[num] == EATING) {//吃饭时间已到 ，开始思考if (clock() - start_eating_time[num] >= eating_time[num]) {state[num] = THINKING;//printf("%d is thinking\n",num);printf("state: %d %d %d %d %d\n",state[0],state[1],state[2],state[3],state[4]);start_thinking_time[num] = clock();  //记录开始思考时间thinking_time[num] = (rand() % 10 + 10) * U_SECOND;  //随机生成思考时间}} else if (state[num] == THINKING) {//思考一定时间后，哲学家饿了，需要吃饭if (clock() - start_thinking_time[num] >= thinking_time[num]) {state[num] = HUNGRY;printf("state: %d %d %d %d %d\n",state[0],state[1],state[2],state[3],state[4]);// printf("%d is hungry\n",num);}}pthread_mutex_unlock(&mutex);       }
}

方法二：

通过互斥信号量 mutex 对哲学家进餐之前取左侧和右侧筷子的操作进行保护，可以防止死锁的出现。

# include <stdio.h>
# include <stdlib.h>
# include <malloc.h>
# include <time.h>
# include <unistd.h>
# include <pthread.h>
# include <semaphore.h># define N 5sem_t chopsticks[N];    //设置5种信号量，有5种不同类型的资源，每一种有1个,这样便于理解，因为每个哲学家需要的资源不同pthread_mutex_t mutex;    //定义互斥锁int philosophers[N] = {0, 1, 2, 3, 4};    //代表5个哲学家的编号void delay (int len) {int i = rand() % len;int x;while (i > 0) {x = rand() % len;while (x > 0) {x--;}i--;}
}void *philosopher (void* arg) {int i = *(int *)arg;int left = i;//左筷子的编号和哲学家的编号相同int right = (i + 1) % N;//右筷子的编号为哲学家编号+1while (1) {printf("哲学家%d正在思考问题\n", i);delay(60000);printf("哲学家%d饿了\n", i);pthread_mutex_lock(&mutex);//加锁sem_wait(&chopsticks[left]);//此时这个哲学家左筷子的信号量-1之后>=0时，表示能继续执行。printf("哲学家%d拿起了%d号筷子,现在只有一支筷子,不能进餐\n", i, left);sem_wait(&chopsticks[right]);printf("哲学家%d拿起了%d号筷子\n", i, right);pthread_mutex_unlock(&mutex);//解锁printf("哲学家%d现在有两支筷子,开始进餐\n", i);delay(60000);sem_post(&chopsticks[left]);printf("哲学家%d放下了%d号筷子\n", i, left);sem_post(&chopsticks[right]);printf("哲学家%d放下了%d号筷子\n", i, right);}
}int main (int argc, char **argv) {srand(time(NULL));pthread_t philo[N];//信号量初始化for (int i=0; i<N; i++) {sem_init(&chopsticks[i], 0, 1);}pthread_mutex_init(&mutex,NULL);//初始化互斥锁//创建线程for (int i=0; i<N; i++) {pthread_create(&philo[i], NULL, philosopher, &philosophers[i]);}//挂起线程for (int i=0; i<N; i++) {pthread_join(philo[i], NULL);}//销毁信号量for (int i=0; i<N; i++) {sem_destroy(&chopsticks[i]);}pthread_mutex_destroy(&mutex);//销毁互斥锁return 0;
}# include <stdio.h>
# include <stdlib.h>
# include <malloc.h>
# include <time.h>
# include <unistd.h>
# include <pthread.h>
# include <semaphore.h># define N 5sem_t chopsticks[N];    //设置5种信号量，有5种不同类型的资源，每一种有1个,这样便于理解，因为每个哲学家需要的资源不同pthread_mutex_t mutex;    //定义互斥锁int philosophers[N] = {0, 1, 2, 3, 4};    //代表5个哲学家的编号void delay (int len) {int i = rand() % len;int x;while (i > 0) {x = rand() % len;while (x > 0) {x--;}i--;}
}void *philosopher (void* arg) {int i = *(int *)arg;int left = i;//左筷子的编号和哲学家的编号相同int right = (i + 1) % N;//右筷子的编号为哲学家编号+1while (1) {printf("哲学家%d正在思考问题\n", i);delay(60000);printf("哲学家%d饿了\n", i);pthread_mutex_lock(&mutex);//加锁sem_wait(&chopsticks[left]);//此时这个哲学家左筷子的信号量-1之后>=0时，表示能继续执行。printf("哲学家%d拿起了%d号筷子,现在只有一支筷子,不能进餐\n", i, left);sem_wait(&chopsticks[right]);printf("哲学家%d拿起了%d号筷子\n", i, right);pthread_mutex_unlock(&mutex);//解锁printf("哲学家%d现在有两支筷子,开始进餐\n", i);delay(60000);sem_post(&chopsticks[left]);printf("哲学家%d放下了%d号筷子\n", i, left);sem_post(&chopsticks[right]);printf("哲学家%d放下了%d号筷子\n", i, right);}
}int main (int argc, char **argv) {srand(time(NULL));pthread_t philo[N];//信号量初始化for (int i=0; i<N; i++) {sem_init(&chopsticks[i], 0, 1);}pthread_mutex_init(&mutex,NULL);//初始化互斥锁//创建线程for (int i=0; i<N; i++) {pthread_create(&philo[i], NULL, philosopher, &philosophers[i]);}//挂起线程for (int i=0; i<N; i++) {pthread_join(philo[i], NULL);}//销毁信号量for (int i=0; i<N; i++) {sem_destroy(&chopsticks[i]);}pthread_mutex_destroy(&mutex);//销毁互斥锁return 0;
}

线程池实现

# include <stdio.h>
# include <stdlib.h>
# include <pthread.h>
# include <errno.h>
# include <sys/types.h>
# include <fcntl.h>
# include <string.h>
# include <time.h>
# include <signal.h>
# include <sys/wait.h>void *thread_routine();
void create(int num);
void add(void(*routine)(void *), void* arg);
void destroy();
void func1();typedef void Func(void*);typedef struct th_queue {void*             arg;Func*             routine;            struct th_queue  *next;
} thpool_queue;typedef struct threadpool {    //线程池int               flag;             // 0不销毁　　1销毁int               num;              // 线程池大小pthread_t         *ID;pthread_cond_t    cond;             // 条件变量pthread_mutex_t   mutex;            // 互斥锁thpool_queue     *head;             // 指向队列指针
} thpool;static thpool * pool = NULL;int main(int argc,char *argv[])
{system("clear");int i;create(10);                             //线程池里创建10个线程for (i = 0; i < 20; i++) {add(&func1,NULL);}sleep(5);destroy();
}void *thread() {                       //线程函数thpool_queue *run;while (1) {pthread_mutex_lock(&pool->mutex);              while((pool->head == NULL) && ( pool->flag == 0)) {    //此时没有任务且不销毁线程池pthread_cond_wait(&pool->cond,&pool->mutex);       //抢到锁的线程等待，其他线程在锁外边阻塞}if(pool->flag != 0) {pthread_mutex_unlock(&pool->mutex);pthread_exit(0);}run = pool->head;                      //不销毁，将任务添加到队列pool->head = pool->head->next;          //让任务指针指向下一个pthread_mutex_unlock(&pool->mutex);run->routine(run->arg);free(run);}
}void create(int num) {                  //线程创建函数int i;   pool = (thpool *)malloc(sizeof(thpool));    if( !pool )perror("malloc error!");pool->flag = 0;pool->num = num;                pool->ID = (pthread_t*)malloc(num*sizeof(pthread_t));pool->head = NULL;pthread_mutex_init(&pool->mutex,NULL);pthread_cond_init(&pool->cond,NULL);for (i = 0;i < num;i++) {pthread_create(&pool->ID[i],NULL,thread,NULL);  //创建线程}
}void add(void(*func)(void *), void* arg) {thpool_queue  *run,*task;run = (thpool_queue*)malloc(sizeof(thpool_queue));run->routine = func;run->arg = arg;run->next = NULL;pthread_mutex_lock(&pool->mutex);               //对队列操作保证只有一个线程task = pool->head;if( !task ) {pool->head = run;                          //任务是第一个任务} else {while (task->next != NULL)                //不是第一个添加到最后task = task->next;task->next = run;}pthread_cond_signal(&pool->cond);pthread_mutex_unlock(&pool->mutex);
}void destroy() {                           //销毁线程池printf("The threadpool is being destroyed!\n");int i;thpool_queue *task;  if(pool->flag != 0)                        //先判断是否已经销毁return;pool->flag = 1;                            //1，则销毁pthread_mutex_lock(&pool->mutex); pthread_cond_broadcast(&pool->cond);       //唤醒全部线程pthread_mutex_unlock(&pool->mutex);   for (i = 0; i < pool->num; i++) {pthread_join(pool->ID[i],NULL);     //等待所有线程都结束}free(pool->ID);while (pool->head) {task = pool->head;pool->head = pool->head->next;free(task);                              //释放每一个任务}pthread_mutex_destroy(&pool->mutex);    //销毁锁pthread_cond_destroy(&pool->cond);      //销毁条件变量free(pool);                         sleep(5);puts("Destroyed!\n");
}void func1() {printf("thread %u is running\n",pthread_self());sleep(3);
}}