Linux应用开发 - 线程同步

线程同步的三种方法

• 互斥锁（Mutex）
• 条件变量（cond）
• 信号量（sem）

互斥锁（Mutex）

存在的意义：用来同步同一个进程中的各个线程。如果这个互斥量存放在多个进程共享的某个内存中，则还能通过互斥量来进行进程间同步。

互斥量是一种特殊的二值信号量，类似于当只有一个车位的停车场，当一辆车进入的时候，将停车场大门锁上，如果再有车辆想进来的话就必须等到里面那辆车出来，才能进去。

声明的数据类型为：pthread_mutex_t 在中有具体定义。

pthread_mutex_t myMutex;

• 互斥锁的初始化

  // ①使用特定的宏
  pthread_mutex_t myMutex = PTHREAD_MUTEX_INITIALIZER;
  // ②调用初始化函数
  pthread_mutex_t myMutex;
  pthread_mutex_init(&myMutex , NULL);
  

  • pthread_mutex_init()函数可以自定义互斥锁的属性

    int pthread_mutex_init(pthread_mutex_t *mutex, const pthread_mutexattr_t *attr);
    

    • mutex：表示要初始化的互斥锁
    • attr：自定义新建互斥锁的属性，为NULL时表示以默认属性创建互斥锁
    • 返回值：成功：0；失败：非零数

• 互斥锁的 “加锁” 和 “解锁”

  • int pthread_mutex_lock(pthread_mutex_t* mutex);   //实现加锁
    int pthread_mutex_trylock(pthread_mutex_t* mutex);  //实现加锁
    int pthread_mutex_unlock(pthread_mutex_t* mutex);   //实现解锁
    

  • pthread_mutex_unlock()函数用于对指定互斥锁进行“解锁操作”

  • pthread_mutex_lock() 和 pthread_mutex_trylock() 函数都用于实现“加锁”操作，不同之处在于当互斥锁已经处于“加锁”状态时 执行 pthread_mutex_lock() 函数会使线程进入等待（阻塞）状态，直至互斥锁得到释放；执行 pthread_mutex_trylock() 函数不会阻塞线程，直接返回非零数（表示加锁失败）。

• 互斥锁的销毁

  • int pthread_mutex_destroy(pthread_mutex_t *mutex);
    

    • mutex：要销毁的互斥锁
    • 返回值：成功：0；失败：非零数

    注意，对于用 PTHREAD_MUTEX_INITIALIZER 或者 pthread_mutex_init() 函数直接初始化的互斥锁，无需调用 pthread_mutex_destory() 函数手动销毁。

案例

/**多线程临界资源访问* 多个线程操作公共资源，如果是全局变量，会出现“矛盾”的现象* 例如：线程1想要变量自增，线程2想要变量自减* 通过互斥量管理共享资源*/
#define _GNU_SOURCE
#include 
#include 
#include 
#include pthread_mutex_t mutex;int num = 0;void *func()
{pthread_mutex_lock(&mutex);while (num < 3){num++;printf("%s num = %d\n", __FUNCTION__, num);sleep(1);/* code */}pthread_mutex_unlock(&mutex);pthread_exit(NULL);
}
void *func1()
{pthread_mutex_lock(&mutex);while (num > -3){num--;printf("%s num = %d\n", __FUNCTION__, num);sleep(1);/* code */}pthread_mutex_unlock(&mutex);pthread_exit(NULL);
}int main()
{pthread_t tid1, tid2;int rmutex = pthread_mutex_init(&mutex,NULL);if (rmutex != 0){perror("init mutex fail\n");return -1;/* code */}int ret = pthread_create(&tid1, NULL, func, NULL);if (ret != 0){perror("creat thread fail\n");return -1;/* code */}ret = pthread_create(&tid2, NULL, func1, NULL);if (ret != 0){perror("creat thread fail\n");return -1;/* code */}pthread_join(tid1, NULL);pthread_join(tid2, NULL);return 0;
}

运行结果：

func num = 1
func num = 2
func num = 3
func1 num = 2
func1 num = 1
func1 num = 0
func1 num = -1
func1 num = -2
func1 num = -3

条件变量（cond）

mutex体现的是一种竞争，我离开了，通知你进来。

cond体现的是一种协作，我准备好了，通知你开始吧。

条件变量是另一种线程同步机制，主要是用来等某个条件发生。可以用来同步同一进程中的各个线程。跟互斥量一样如果存放在多个进程共享的某个内存中时，可以通过条件来进行进程间的同步。

每个条件变量总是和一个互斥量关联，条件本身是由互斥量保护的，线程在改变条件状态之间必须要锁住互斥量。

条件变量相对于互斥量最大的优点是允许线程以无竞争的方式灯等待条件的发生。当某个线程获取到互斥锁之后，发现需要等待某个条件变量为真，如果是这样，该线程就可以等待在某个条件上，不需要通过轮训的方式占用CPU

案例

/**条件变量和互斥量*/
#define _GNU_SOURCE
#include 
#include 
#include 
#include #define     BUFFER_SIZE     5   //产品库存大小
#define     PRODUCT_CNT     6  //产品生产总数struct product_cons
{int buffer[BUFFER_SIZE];pthread_mutex_t lock;   //互斥锁int readpos,writepos;pthread_cond_t  notempty;   //条件变量非空pthread_cond_t  notfull;    //条件变量非满/* data */
}buffer;void init(struct product_cons *p)
{pthread_mutex_init(&p->lock,NULL);pthread_cond_init(&p->notempty,NULL);pthread_cond_init(&p->notfull,NULL);p->readpos = 0;p->writepos = 0;
}void finish(struct product_cons *p)
{pthread_mutex_destroy(&p->lock);pthread_cond_destroy(&p->notempty);pthread_cond_destroy(&p->notfull);p->readpos = 0;p->writepos = 0;
}void put(struct product_cons *p,int data)
{pthread_mutex_lock(&p->lock);if((p->writepos+1)%BUFFER_SIZE == p->readpos)//检查仓库非满状态才能生产{printf("curr full\n");pthread_cond_wait(&p->notfull,&p->lock);}p->buffer[p->writepos] = data;p->writepos++;if(p->writepos >= BUFFER_SIZE)p->writepos = 0;pthread_cond_signal(&p->notempty);pthread_mutex_unlock(&p->lock);
}int get(struct product_cons *p)
{int data;pthread_mutex_lock(&p->lock);if(p->readpos == p->writepos){printf("consumer wait not empty\n");pthread_cond_wait(&p->notempty,&p->lock);}data = p->buffer[p->readpos];p->readpos++;if(p->readpos>=BUFFER_SIZE)p->readpos = 0;pthread_cond_signal(&p->notfull);pthread_mutex_unlock(&p->lock);return data;
}void *producer(void *data) //子线程，生产
{for (int i = 0; i < 50; i++)    //生产50个产品{sleep(1);printf("put the %d product..\n",i);put(&buffer,i);// printf("put the %d product success\n",i);/* code */}printf("producer stopped\n");return NULL;
}void *consumer(void *data) //子线程，消费
{static int cnt = 0;int num;while (1){sleep(2);// printf("get product...\n");num = get(&buffer);printf("get the %d product success\n",num);if(++cnt == PRODUCT_CNT)break;/* code */}printf("consumer stopped\n");return  NULL;
}int main()
{pthread_t tid1, tid2;init(&buffer);int ret = pthread_create(&tid1, NULL, producer, NULL);if (ret != 0){perror("creat thread fail\n");return -1;/* code */}ret = pthread_create(&tid2, NULL, consumer, NULL);if (ret != 0){perror("creat thread fail\n");return -1;/* code */}pthread_join(tid1, NULL);pthread_join(tid2, NULL);finish(&buffer);return 0;
}

运行结果：

put the 0 product..
get the 0 product success
put the 1 product..
put the 2 product..
get the 1 product success
put the 3 product..
put the 4 product..
get the 2 product success
put the 5 product..
put the 6 product..
get the 3 product success
put the 7 product..
put the 8 product..
curr full
get the 4 product success
put the 9 product..
curr full
get the 5 product success
consumer stopped
put the 10 product..
curr full

这个例程主要就是解决生产者和消费者冲突的问题，互斥量的话也能实现，只不过生产完了再消费。加个条件变量的话，边生产边消费。

信号量（sem）

如同进程一样，线程也可以通过信号量来实现通信，虽然是轻量级的。信号量函数的名字都以"sem_"打头。线程使用的基本信号量函数有四个。

• 信号量初始化

  int sem_init (sem_t *sem , int pshared, unsigned int value);

  • 这是对由sem指定的信号量进行初始化，设置好它的共享选项(linux 只支持为0，即表示它是当前进程的局部信号量)，然后给它一个初始值VALUE

  int sem_wait(sem_t *sem);

  • 等待信号量。给信号量减1，然后等待直到信号量的值大于0。

  int sem_post(sem_t *sem);

  • 释放信号量。信号量值加1。并通知其他等待线程。

  int sem_destroy(sem_t *sem);

  • 销毁信号量。我们用完信号量后都它进行清理。归还占有的一切资源。

案例

/**线程信号量控制测试**/
#define _GNU_SOURCE
#include 
#include 
#include 
#include 
#include sem_t sem_1, sem_2, sem_3;void *func1()
{sem_wait(&sem_1);printf("%s,come!\n", __FUNCTION__);sem_post(&sem_2);pthread_exit(NULL);
}
void *func2()
{sem_wait(&sem_2);printf("%s,come!\n", __FUNCTION__);sem_post(&sem_3);pthread_exit(NULL);
}
void *func3()
{sem_wait(&sem_3);printf("%s,come!\n", __FUNCTION__);sem_post(&sem_1);pthread_exit(NULL);
}int main()
{pthread_t tid1, tid2, tid3;/*int sem_init(sem_t *sem,int pshared,unsigned int value);参数1：sem_t类型地址参数2：0表示线程控制，否则为进程控制参数3：信号量的初始值0：阻塞1：运行*/int rsem = sem_init(&sem_1, 0, 1);if (rsem < 0){perror("sem_init fail\n");return -1;/* code */}rsem = sem_init(&sem_2, 0, 0);if (rsem < 0){perror("sem_init fail\n");return -1;/* code */}rsem = sem_init(&sem_3, 0, 0);if (rsem < 0){perror("sem_init fail\n");return -1;/* code */}int ret = pthread_create(&tid1, NULL, func1, NULL);if (ret != 0){perror("creat thread fail\n");return -1;/* code */}ret = pthread_create(&tid2, NULL, func2, NULL);if (ret != 0){perror("creat thread fail\n");return -1;/* code */}ret = pthread_create(&tid3, NULL, func3, NULL);if (ret != 0){perror("creat thread fail\n");return -1;/* code */}/*回收线程资源*/pthread_join(tid1, NULL);pthread_join(tid2, NULL);pthread_join(tid3, NULL);/*销毁信号量*/sem_destroy(&sem_1);sem_destroy(&sem_2);sem_destroy(&sem_3);return 0;
}

运行效果：

func1,come!
func2,come!
func3,come!

在这个案例中，只初始化sem_1值为1，sem_2 sem_3初始化都为0，线程1中sem_post给信号量2控制块，这样线程2就能运行。