pthread线程的同步问题

1 互斥变量： pthread_mutex_t，本质上说是一把锁，在访问共享资源前对互斥量加锁，访问完成后释放锁。

   对互斥量进行加锁后，任何其他试图再次对互斥量加锁的线程将会被阻塞直至当前线程释放该互斥锁。

pthread_mutex_lock : 加锁

pthread_mutex_unlock：解锁

2 条件变量：pthread_cond_t，多个线程协作的一种同步方式，若某个条件不满足，线程将一直等待。条件变量一般

和互斥变量一起配合使用，因为条件变量本身是互斥访问的，所以要有一个互斥变量保护它。

pthread_cond_wait: 若条件不满组，一直等待下去

pthread_cond_signal: 把条件变量设为满足状态

3 实例： 两个线程，共同操作消息队列，线程1不断产生消息，添加到消息队列中，线程2不断处理消息队列中的消息

若消息队列为空，则等待直到有新的消息。写一个程序模拟此过程。

设置一个条件变量, pthread_cond_t    qready， 表示消息队列不为空

设置一个互斥变量, pthread_mutex_t  mutex， 用来保护qready和消息队列的互斥访问

代码如下：

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

#define SIZE 10000

typedef struct msg_t {
	int				data;
	struct msg_t 	*next;
} msg_t;

typedef struct global_t {
	msg_t 				*workq;
	pthread_cond_t 		qready;
	pthread_mutex_t		qlock;
	int					size;	/* 队列中待处理的消息数量 */
} global_t;

void *process_msg(void *arg)
{
	msg_t 		*mp;
	global_t	*global;
	int			done;

	global = (global_t *)arg;
	done = 0;
	while (done < SIZE) {
		pthread_mutex_lock(&global->qlock);
		while (global->size < 1) {
			printf("开始等待##############\n");
			pthread_cond_wait(&global->qready, &global->qlock);
			printf("等待结束##############\n");
		}
		mp = global->workq;
		global->workq = mp->next;
		global->size--;
		pthread_mutex_unlock(&global->qlock);
		printf("处理%d\n", mp->data);
		done++;
	}
	return (void *)0;
}

void enqueue_msg(global_t *global, msg_t *mp)
{
	pthread_mutex_lock(&global->qlock);
	mp->next = global->workq;
	global->workq = mp;
	global->size++;
	pthread_mutex_unlock(&global->qlock);
	pthread_cond_signal(&global->qready);
}

void *generate_msg(void *arg)
{
	int			i;
	global_t	*global;
	msg_t		*msg;

	global = (global_t *)arg;
	for (i = 0; i < SIZE; i++) {
		msg = malloc(sizeof(msg_t));
		msg->data = i;
		printf("%d入队...\n", i);
		enqueue_msg(global, msg);
		printf("%d入队完成...\n", i);
	}

	return (void *)0;
}

int main()
{
	pthread_t	ptid, gtid;
	global_t	global;
	void		*tret;

	global.workq = NULL;
	global.size = 0;

	pthread_cond_init(&global.qready, NULL);
	pthread_mutex_init(&global.qlock, NULL);

	pthread_create(&ptid, NULL, process_msg, &global);
	pthread_create(&gtid, NULL, generate_msg, &global);

	pthread_join(ptid, &tret);
	pthread_join(gtid, &tret);

	return 0;
}

 

上面的例子开启了两个线程分别用来产生和消耗消息，加上main，总共有3个线程在跑，如果把产生消息的工作放到

 main函数中，是否依然需要同步呢？当然需要，因为main函数本身也是一个线程，依然要和处理消息的线程保持同步。

 代码如下：

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

#define SIZE 10000

typedef struct msg_t {
	int				data;
	struct msg_t 	*next;
} msg_t;

typedef struct global_t {
	msg_t 				*workq;
	pthread_cond_t 		qready;
	pthread_mutex_t		qlock;
	int					size;	/* 队列中待处理的消息数量 */
} global_t;

void *process_msg(void *arg)
{
	msg_t 		*mp;
	global_t	*global;
	int			done;

	global = (global_t *)arg;
	done = 0;
	while (done < SIZE) {
		pthread_mutex_lock(&global->qlock);
		while (global->size < 1) {
			printf("开始等待##############\n");
			pthread_cond_wait(&global->qready, &global->qlock);
			printf("等待结束##############\n");
		}
		mp = global->workq;
		global->workq = mp->next;
		global->size--;
		pthread_mutex_unlock(&global->qlock);
		printf("处理%d\n", mp->data);
		done++;
	}
	return (void *)0;
}

void enqueue_msg(global_t *global, msg_t *mp)
{
	pthread_mutex_lock(&global->qlock);
	mp->next = global->workq;
	global->workq = mp;
	global->size++;
	pthread_mutex_unlock(&global->qlock);
	pthread_cond_signal(&global->qready);
}

void *generate_msg(void *arg)
{
	int			i;
	global_t	*global;
	msg_t		*msg;

	global = (global_t *)arg;
	for (i = 0; i < SIZE; i++) {
		msg = malloc(sizeof(msg_t));
		msg->data = i;
		printf("%d入队...\n", i);
		enqueue_msg(global, msg);
		printf("%d入队完成...\n", i);
	}

	return (void *)0;
}

int main()
{
	pthread_t	ptid, gtid;
	global_t	global;
	void		*tret;

	global.workq = NULL;
	global.size = 0;

	pthread_cond_init(&global.qready, NULL);
	pthread_mutex_init(&global.qlock, NULL);

	pthread_create(&ptid, NULL, process_msg, &global);
	generate_msg(&global);		/* 把产生消息的工作放到main函数中 */

	pthread_join(ptid, &tret);

	return 0;
}

 

4 信号量 semaphore

mutex变量是非0即1的，可看作一种资源的可用数量，初始化时mutex是1，表示有一个可用资源，加锁

时获得该资源，将mutex减到0，表示不再有可用资源，解锁时释放该资源，将mutex重新加到1，表示又有了

一个可用资源。信号量（semaphore）和mutex类似，表示可用资源的数量，和mutex不同的是这个数量可以大于1。

这种信号量不仅可用于同一进程的线程间同步，也可用于不同进程间的同步。

linux中提供的接口为：

#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_destroy(sem_t * sem);

 

semaphore变量的类型为sem_t，sem_init()初始化一个semaphore变量，value参数表示可用资源的数量，

pshared参数为0表示信号量用于同一进程的线程间同步。在用完semaphore变量之后应该

调用sem_destroy()释放与semaphore相关的资源。

调用sem_wait()可以获得资源，使semaphore的值减1，如果调用sem_wait()时semaphore的值已经是0，

则挂起等待。如果不希望挂起等待，可以调用sem_trywait()。调用sem_post()可以释放资源，

使semaphore的值加1，同时唤醒挂起等待的线程。

semaphore实例：生产者和消费者问题

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

#define NUM 5
int queue[NUM];
sem_t blank_number, product_number;

void *producer(void *arg) 
{
	int p = 0;
	while (1) {
		sem_wait(&blank_number);
		queue[p] = rand() % 1000 + 1;
		printf("Produce %d\n", queue[p]);
		sem_post(&product_number);
		p = (p+1)%NUM;
		sleep(rand()%5);
	}
}

void *consumer(void *arg) 
{
	int c = 0;
	while (1) {
		sem_wait(&product_number);
		printf("Consume %d\n", queue[c]);
		queue[c] = 0;
		sem_post(&blank_number);
		c = (c+1)%NUM;
		sleep(rand()%5);
	}
}

int main(int argc, char *argv[]) 
{
	pthread_t pid, cid;  

	sem_init(&blank_number, 0, NUM);
	sem_init(&product_number, 0, 0);
	pthread_create(&pid, NULL, producer, NULL);
	pthread_create(&cid, NULL, consumer, NULL);
	pthread_join(pid, NULL);
	pthread_join(cid, NULL);
	sem_destroy(&blank_number);
	sem_destroy(&product_number);
	return 0;
}