线程同步-生产者消费者问题

#include <stdio.h>
#include <pthread.h>

#define LOOP_COUNT 5			//生产者和消费者各自循环次数
pthread_mutex_t mutex;			//定义一个全局互斥量，在不同函数中
								//初始化和使用

void *producer( void *arg );	//生产者线程
void *consumer( void *arg );	//消费者线程

int main(int argc , char *argv[]){
	pthread_t thrd_prod , thrd_cons;

	pthread_mutex_init( &mutex , NULL );	//初始化互斥量

	//创建生产者和消费者线程
	if( pthread_create( &thrd_prod , NULL, producer ,
				NULL ) != 0 )
		oops( "thread create failed." );
	sleep(1);								//保证生产者线程先运行

	if( pthread_create( &thrd_cons , NULL, consumer ,
				NULL ) != 0 )
		oops( "thread create failed." );

	//等待线程结束
	if( pthread_join( thrd_prod , NULL ) != 0 )
		oops( " wait thread failed.");
	if( pthread_join( thrd_cons , NULL ) != 0 )
		oops( " wait thread failed.");

	pthread_mutex_destroy( &mutex );		//关闭互斥量
	return 0;
}

void *producer( void *arg){
	int count = 0 ;				//循环计数

	while( count++ < LOOP_COUNT ){
		pthread_mutex_lock( &mutex );	//加锁

		//成功占有互斥量，接下来可以对缓冲区（仓库）进行生产
		//操作
		printf( " producer put a product to buffer.\n");
		sleep(3);				//休眠3秒， 便于程序观察

		pthread_mutex_unlock( &mutex ); //解锁
		sleep(1);				//休眠一秒，防止它又马上占据锁
	}
}
void *consumer( void *arg ){
	int count = 0 ;				//循环计数

	while( count++ < LOOP_COUNT ){
//		sleep(2);				//休眠一秒， 便于程序观察
		pthread_mutex_lock( &mutex );	//加锁

		//成功占有互斥量，接下来可以对缓冲区（仓库）进行取出
		//操作
		printf( " consumer get a product from buffer.\n");

		pthread_mutex_unlock( &mutex ); //解锁
		sleep(1);				//休眠一秒，防止它又马上占据锁
	}
}

二、读写锁

#include <stdio.h>
#include <pthread.h>

#define LOOP_COUNT 2			//生产者和消费者各自循环次数
#define LOOP_THRD 5				//消费者线程个数
pthread_rwlock_t rwlock;		//定义一个全局读写锁，在不同函数中
								//初始化和使用

void *producer( void *arg );	//生产者线程
void *consumer( void *arg );	//消费者线程

int main(int argc , char *argv[]){
	int thrd_num ,thrd_id[LOOP_THRD]  ;
	pthread_t thrd_prod , thrd_cons[LOOP_THRD];

	pthread_rwlock_init( &rwlock , NULL );	//初始化互斥量

	//创建一个生产者和多个消费者线程
	if( pthread_create( &thrd_prod , NULL, producer ,
				NULL ) != 0 )
		oops( "thread create failed." );

	for( thrd_num = 0 ; thrd_num < LOOP_THRD; thrd_num++ ){
		thrd_id[thrd_num] = thrd_num;		//线程id，注意线程共享变量
		if( pthread_create( &thrd_cons[thrd_num], NULL,	consumer 
					, <span style="background-color: rgb(255, 0, 0);">(void *)( thrd_id+thrd_num)</span> ) != 0 )
			oops( "thread %d create failed." , thrd_num );
	}

	//等待线程结束
	if( pthread_join( thrd_prod , NULL ) != 0 )
		oops( " wait thread failed.");
	for( thrd_num = 0 ; thrd_num < LOOP_THRD; thrd_num++ ){
		if( pthread_join( thrd_cons[thrd_num] , NULL ) != 0 )
			oops( " wait thread %d failed." , thrd_num);
//		printf("wait %d thread.\n" , thrd_num);
	}
	pthread_rwlock_destroy( &rwlock );		//关闭互斥量
	return 0;
}

void *producer( void *arg){
	int count = 0 ;				//循环计数

	while( count++ < LOOP_COUNT ){
		printf( "producer try to lock wrlock.\n");
		pthread_rwlock_wrlock( &rwlock );	//加锁

		//成功占有互斥量，接下来可以对缓冲区（仓库）进行生产
		//操作
		printf( "producer lock successful, producer put a product to buffer.\n");
		
		/*
			休眠3秒， 便于程序观察,可以看到
			其他读取线程不能占据锁而阻塞
		*/		
		sleep(3);				
		printf("prducer finished ,unlock wrlock.\n");
		pthread_rwlock_unlock( &rwlock ); //解锁
		sleep(1);							//休眠一秒， 防止马上又占据写锁
	}
}
void *consumer( void *arg ){
	int count = 0 ;							//循环计数
	int thrd_id = *( ( int*)arg );

//	printf( "consumer %d ,%#x . \n" , thrd_id ,arg);
	while( count++ < LOOP_COUNT ){
//		sleep( thrd_id+1 );					//休眠一秒， 便于程序观察
		printf( "consumer try to lock rdlock.\n" );
		pthread_rwlock_rdlock( &rwlock );	//加锁

		//成功占有互斥量，接下来可以对缓冲区（仓库）进行取出
		//操作
		printf( " consumer locked successful ,consumer %d get a product from buffer."
				"\n" , thrd_id);
		/*
			休眠3秒， 便于程序观察,可以看到
			其他读取线程能占据读锁
		*/
		sleep(3);
		printf("consumer finished ,unlock rdlock.\n");
		pthread_rwlock_unlock( &rwlock );	//解锁
		sleep(thrd_id+1);							//休眠一秒， 防止马上又占据读锁
	}
}

三、条件变量（cond）

与互斥锁不同，条件变量是用来等待而不是用来上锁的。条件变量用来自动阻塞一个线程，直到某特殊情况发生为止。通常条件变量和互斥锁同时使用。条件变量分为两部分: 条件和变量。条件本身是由互斥量保护的。线程在改变条件状态前先要锁住互斥量。条件变量使我们可以睡眠等待某种条件出现。条件变量是利用线程间共享的全局变量进行同步的一种机制，主要包括两个动作：一个线程等待"条件变量的条件成立"而挂起；另一个线程使"条件成立"（给出条件成立信号）。条件的检测是在互斥锁的保护下进行的。如果一个条件为假，一个线程自动阻塞，并释放等待状态改变的互斥锁。如果另一个线程改变了条件，它发信号给关联的条件变量，唤醒一个或多个等待它的线程，重新获得互斥锁，重新评价条件。如果两进程共享可读写的内存，条件变量可以被用来实现这两进程间的线程同步。

1. 1.初始化条件变量。
   静态态初始化，pthread_cond_t cond = PTHREAD_COND_INITIALIER;
   动态初始化，int pthread_cond_init(pthread_cond_t *cond, pthread_condattr_t *cond_attr);
   
2. 2.等待条件成立。释放锁,同时阻塞等待条件变量为真才行。timewait()设置等待时间,仍未signal,返回ETIMEOUT(加锁保证只有一个线程wait)
   int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex);
   int pthread_cond_timewait(pthread_cond_t *cond,pthread_mutex *mutex,const timespec *abstime);
   
3. 3.激活条件变量。pthread_cond_signal,pthread_cond_broadcast（激活所有等待线程）
   int pthread_cond_signal(pthread_cond_t *cond);
   int pthread_cond_broadcast(pthread_cond_t *cond); //解除所有线程的阻塞
   
4. 4.清除条件变量。无线程等待,否则返回EBUSY
   int pthread_cond_destroy(pthread_cond_t *cond);
5. 接下来我们又改变一下生产者消费者问题：现在缓冲区或者仓库大小为BUFSIZE，只有一个生产者和一个消费者（其实也适用于多个生产者和消费者）， 我们这个时候就可以把缓冲区设置为一个互斥量，一次要么生产者要么消费者霸占它。但接下来处理方式与互斥量有所不同：假如生产者成功占据锁（缓冲区），这时它不能马上开始往里面生产东西， 要先判断缓冲区是不是满的，如果缓冲区满了，那么生产者就会把自己放到等待条件的线程列表上，然后对互斥量进行解锁，这是一个原子操作。如果缓冲区不满则可以生产产品，然后给消费者发送notempty信号，表示缓冲区有产品了， 你可以yy了。然后解锁互斥量。假如是消费者成功占据锁（缓冲区），同样它要检查缓冲区是不是空的，如果空，那么消费者就会把自己放到等待条件的线程列表上，然后对互斥量进行解锁。如果不空，消费者开始yy，然后给生产者发送nofull信号， 表示缓冲区有位置可以生产了， 你快生产吧。然后解锁互斥量。就这样， 生产者消费者和谐同步工作着。
6. 流程图我就不画了，看代码也能明白过程：
7. ---producer过程:lock(mutex)->check notfull->(if notfull wait until notfull)->produce product->send notempty to consumer->unlock(mutex)
8. ---consumer过程：lock(mutex)->check notempty->(if notempty wait until notempty)->get product from buffer->send notfull to poducer->unlock(mutex)

#include <stdio.h>
#include <pthread.h>

#define LOOP_COUNT 20				//生产者和消费者各自循环次数,也可以说生产商品的总量
//#define LOOP_THRD 5				//消费者线程个数
#define BUFSIZE 5					//缓冲区大小，也就是最多能放多少个产品

pthread_mutex_t mutex;				//定义一个全局互斥量，在不同函数中
									//初始化和使用
pthread_cond_t notempty , notfull;	//定义两个条件变量，当作信号投放
unsigned int prod_pos = 3;			//定义生产者在缓冲区开始生产的位置
unsigned int cons_pos = 0;			//定义消费者在缓冲区开始消费的位置

void *producer( void *arg );		//生产者线程
void *consumer( void *arg );		//消费者线程

int main(int argc , char *argv[]){
	pthread_t thrd_prod , thrd_cons;

	pthread_mutex_init( &mutex , NULL );	//初始化互斥量

	//创建生产者和消费者线程
	if( pthread_create( &thrd_prod , NULL, producer ,
				NULL ) != 0 )
		oops( "thread create failed." );
	sleep(1);								//保证生产者线程先运行

	if( pthread_create( &thrd_cons , NULL, consumer ,
				NULL ) != 0 )
		oops( "thread create failed." );

	//等待线程结束
	if( pthread_join( thrd_prod , NULL ) != 0 )
		oops( " wait thread failed.");
	if( pthread_join( thrd_cons , NULL ) != 0 )
		oops( " wait thread failed.");

	pthread_mutex_destroy( &mutex );		//关闭互斥量
	return 0;
}

void *producer( void *arg){
	int count = 0 ;				//循环计数

	while( count++ < LOOP_COUNT ){
		printf( "producer try to lock .\n");
		pthread_mutex_lock( &mutex );	//加锁

		/*
		   成功占有互斥量，接着检查缓冲区是不是满了，
		*/
		if( ( prod_pos + 1 ) % BUFSIZE == cons_pos ){
			//缓冲区满了
			printf( "producer wait not full.\n");
			pthread_cond_wait( &notfull , &mutex );	//等待条件满足
		}
		//如果没满，接下来可以对缓冲区（仓库）进行生产
		//操作
		printf( "producer lock successful, producer put %d's "
				"product to buffer.\n" ,count);
		prod_pos = ( prod_pos +1 ) % BUFSIZE;		//下标前进一个
		pthread_cond_signal( &notempty );			//向消费者发送信号
		/*
			休眠3秒， 便于程序观察,可以看到
			其他读取线程不能占据锁而阻塞
		*/		
		sleep( 1 );				
		printf("prducer finished ,unlock lock.\n");
		pthread_mutex_unlock( &mutex ); //解锁
		sleep( 1 );							//休眠一秒， 防止马上又占据写锁
	}
}
void *consumer( void *arg ){
	int count = 0 ;							//循环计数

	while( count++ < LOOP_COUNT ){
//		sleep( thrd_id+1 );					//休眠一秒， 便于程序观察
		printf( "consumer try to lock .\n" );
		pthread_mutex_lock( &mutex ); //解锁

		/*
		   成功占有互斥量，接下来检查缓冲区是否为空
		*/
		if( cons_pos == prod_pos ){
			printf( "consumer wait not empty.\n");
			pthread_cond_wait( &notempty , &mutex );
		}

		//缓冲区不空，可以对缓冲区（仓库）进行取出操作
		printf( " consumer locked successful ,consumer  "
				"get %d product from buffer.\n" , count);
		cons_pos = ( cons_pos + 1) % BUFSIZE ;	//下标前进一个
		pthread_cond_signal( &notfull );		//向生产着发送信号

		/*
			休眠3秒， 便于程序观察,可以看到
			其他读取线程能占据读锁
		*/
		sleep( 1 );
		printf("consumer finished ,unlock lock.\n");
		pthread_mutex_unlock( &mutex );			//解锁
		sleep(1);						//休眠一秒， 防止马上又占据读锁
	}
}