线程同步：条件变量实现生产者消费者模型

今天药忘吃喽~ 2022-08-19 05:53 187阅读 0赞

**概念：**

假设实现一个生产者消费者模型，仓库只能装100万件item，生产者每次生产1件，消费者每次消费1件，为了保持数据的同步，那么生产者每次都需要检查仓库是否满了，消费者每次都需要检查仓库是否空了，在加上互斥锁前提下，数据是能得同步保证的。循环的访问且每次都给互斥锁加锁解锁，又称为轮询，如果经常做无用功则线程一直在空转，十分浪费CPU。如果在仓库满了，我们希望生产者停下来等待，同理如果仓库空了，我们希望消费者停下来等待。

**互斥锁的初始化：**

如果互斥锁是静态分配的，则可以初始化成常值PTHREAD\_MUTEX\_INITIALIZER，也可以调用pthread\_mutex\_init函数来初始化

如果互斥锁是动态分配的（例如通过malloc）,或者分配在共享内存中，则需要在运行时调用pthread\_mutex\_init函数来初始化

**条件变量初始化：**

如果条件变量是静态分配的，则可以初始化成常值PTHREAD\_COND\_INITIALIZER，也可以调用pthread\_cond\_init函数来初始化

如果条件变量是动态分配的（例如通过malloc）,或者分配在共享内存中，则需要在运行时调用pthread\_cond\_init函数来初始化

**为什么要用while循环而不是if：**

while (nready.nready == 0) {
        Pthread_cond_wait(&nready.cond, &nready.mutex);
    }

pthread\_cond\_signal：点播，通知一个wait线程

pthread\_cond\_broadcast：广播，通知所有的wait线程

1.假设仓库为空，有2个消费者在等待商品，设为C1和C2

2.假设生产者只生产了1件商品，然后调用pthread\_cond\_broadcast，则C1和C2都会得到通知

3.假设C1比C2先得到通知，然后加锁把商品消费了，并且解锁，这时C2就能拿到锁，但是此时商品已经没有了，如果此时C2不做检测，则会出现数据同步问题

**简单的测试代码：**

**----- gcc cond\_prodcons.c -lpthread -----**

#include <pthread.h>
    #include <errno.h>
    #include <stdlib.h>
    #include <stdio.h>
    #include <unistd.h>
    
    #define MAXITEMS 1000000 /* 需要生产n个item */
    #define MAXNTHREADS 10 /* 生产线n条 */
    
    int buff[MAXITEMS];
    
    struct {
    	pthread_mutex_t mutex;
    	int nput;
    	int nval;
    } put = {
    	PTHREAD_MUTEX_INITIALIZER
    };
    
    struct {
    	pthread_mutex_t mutex;
    	pthread_cond_t cond;
    	int nready;
    } nready = {
    	PTHREAD_MUTEX_INITIALIZER, PTHREAD_COND_INITIALIZER
    };
    
    //创建线程
    int Pthread_create(pthread_t *thread, const pthread_attr_t *attr, void *(*start_routine) (void *), void *arg) {
    	int res = pthread_create(thread, attr, start_routine, arg);
    	if (res != 0) {
    		perror("pthread_create fail");
    		exit(EXIT_FAILURE);
    	}	
    }
    
    //互斥锁
    int Pthread_mutex_lock(pthread_mutex_t *mutex) {
    	int res = pthread_mutex_lock(mutex);
    	if (res != 0) {
    		perror("Pthread_mutex_lock fail");
    	}	
    }
    
    //互斥解锁
    int Pthread_mutex_unlock(pthread_mutex_t *mutex) {
    	int res = pthread_mutex_unlock(mutex);
    	if (res != 0) {
    		perror("Pthread_mutex_unlock fail");
    	}	
    }
    
    //发送线程信号
    int Pthread_cond_signal(pthread_cond_t *cond) {
    	int res = pthread_cond_signal(cond);
    	if (res != 0) {
    		perror("Pthread_cond_signal fail");
    	}	
    }
    
    /*
    pthread_cond_wait会释放mutex,并进入阻塞状态,
    一旦获取到信号则会将mutex锁住，并返回
    */
    int Pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex) {
    	int res = pthread_cond_wait(cond, mutex);
    	if (res != 0) {
    		perror("pthread_cond_wait fail");
    	}	
    }
    
    //线程函数声明
    void* produce(void*);
    void* consume(void*);
    
    int main(int argc, char const *argv[])
    {
    	int i, count[MAXNTHREADS];
    	pthread_t tid_produce[MAXNTHREADS], tid_consume;
    
    	// n个生产者
    	for (i=0; i<MAXNTHREADS; ++i) {
    		count[i] = 0;
    		Pthread_create(&tid_produce[i], NULL, produce, &count[i]);
    	}
    
    	// 1个消费者，也可以是n个
    	Pthread_create(&tid_consume, NULL, consume, NULL);
    
    	for (i=0; i<MAXNTHREADS; ++i) {
    		pthread_join(tid_produce[i], NULL);
    		printf("count[%d] = %d\n", i, count[i]);
    	}
    
    	pthread_join(tid_consume, NULL);
    
    	return 0;
    }
    
    // 生产者
    void* produce(void* arg) {
    	
    	int dosignal;
    	for ( ; ; ) {
    		Pthread_mutex_lock(&put.mutex);
    		if (put.nput >= MAXITEMS) {
    			Pthread_mutex_unlock(&put.mutex);
    			return(NULL); /* 数据填满了 */
    		}
    
    		buff[put.nput] = put.nval;
    		put.nput++;
    		put.nval++;
    		Pthread_mutex_unlock(&put.mutex);
    
    		Pthread_mutex_lock(&nready.mutex);
    		dosignal = (nready.nready == 0); /* 是否需要唤醒 */
    		nready.nready++;
    		Pthread_mutex_unlock(&nready.mutex);
    
    		if (dosignal) {
    			Pthread_cond_signal(&nready.cond);
    		}
    
    		*((int *)arg) += 1; /* 当前线程生产了多少个数据 */
    
    	}
    
    	return(NULL);
    }
    
    
    // 消费者
    void* consume(void* arg) {
    
    	int i;
    
    	for (i=0; i<MAXITEMS; i++) {
    
    		Pthread_mutex_lock(&nready.mutex);
    		while (nready.nready == 0) { /* while 循环是防止wait返回时发生虚假唤醒 */
    			Pthread_cond_wait(&nready.cond, &nready.mutex);
    		}
    		nready.nready--;
    		Pthread_mutex_unlock(&nready.mutex);
    
    		//如果线程没同步成功,则会有输出
    		if (buff[i] != i)
    			printf("buff[%d] = %d\n", i, buff[i]);
    	}
    
    	return(NULL);
    }

END