Linux线程同步的三种方法如何组合成一个长尾词？

本文共计3164个文字，预计阅读时间需要13分钟。

程序的显著特点是资源的共享性，但资源共享中的同步问题是多线程编程的难点。Linux下提供了多种方式来处理线程同步，最常用的是互斥锁（mutex）、条件变量和信号量。

一、互斥锁（mutex）互斥锁是一种同步机制，用于确保在同一时刻只有一个线程可以访问共享资源。通过互斥锁，可以避免多个线程同时修改同一资源，从而防止数据竞争。

二、条件变量条件变量是一种线程同步机制，允许线程在某些条件未满足时挂起，直到其他线程改变条件并通知它们。这有助于实现复杂的同步逻辑。

三、信号量信号量是一种更高级的同步机制，可以表示资源的数量。线程可以通过信号量来申请或释放资源，从而实现线程间的同步。

线程的最大特点是资源的共享性，但资源共享中的同步问题是多线程编程的难点。linux下提供了多种方式来处理线程同步，最常用的是互斥锁、条件变量和信号量。

一、互斥锁(mutex)

通过锁机制实现线程间的同步。

1. 初始化锁。在Linux下，线程的互斥量数据类型是pthread_mutex_t。在使用前,要对它进行初始化。
   静态分配：pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
   动态分配：int pthread_mutex_init(pthread_mutex_t *mutex, const pthread_mutex_attr_t *mutexattr);
2. 加锁。对共享资源的访问，要对互斥量进行加锁，如果互斥量已经上了锁，调用线程会阻塞，直到互斥量被解锁。
   int pthread_mutex_lock(pthread_mutex *mutex);
   int pthread_mutex_trylock(pthread_mutex_t *mutex);
3. 解锁。在完成了对共享资源的访问后，要对互斥量进行解锁。
   int pthread_mutex_unlock(pthread_mutex_t *mutex);
4. 销毁锁。锁在是使用完成后，需要进行销毁以释放资源。
   int pthread_mutex_destroy(pthread_mutex *mutex);

二、条件变量(cond)

1. #include <cstdio> 2. #include <cstdlib> 3. #include <unistd.h> 4. #include <pthread.h> 5. #include "iostream" 6. using namespace std; 7. pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; 8. int tmp; 9. void* thread(void *arg) 10. { 11. cout << "thread id is " << pthread_self() << endl; 12. pthread_mutex_lock(&mutex); 13. tmp = 12; 14. cout << "Now a is " << tmp << endl; 15. pthread_mutex_unlock(&mutex); 16. return NULL; 17. } 18. int main() 19. { 20. pthread_t id; 21. cout << "main thread id is " << pthread_self() << endl; 22. tmp = 3; 23. cout << "In main func tmp = " << tmp << endl; 24. if (!pthread_create(&id, NULL, thread, NULL)) 25. { 26. cout << "Create thread success!" << endl; 27. } 28. else 29. { 30. cout << "Create thread failed!" << endl; 31. } 32. pthread_join(id, NULL); 33. pthread_mutex_destroy(&mutex); 34. return 0; 35. } 36. //编译：g++ -o thread testthread.cpp -lpthread

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

1. 初始化条件变量。
   静态态初始化，pthread_cond_t cond = PTHREAD_COND_INITIALIER;
   动态初始化，int pthread_cond_init(pthread_cond_t *cond, pthread_condattr_t *cond_attr);
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. 激活条件变量。pthread_cond_signal,pthread_cond_broadcast（激活所有等待线程）
   int pthread_cond_signal(pthread_cond_t *cond);
   int pthread_cond_broadcast(pthread_cond_t *cond); //解除所有线程的阻塞
4. 清除条件变量。无线程等待,否则返回EBUSY
   int pthread_cond_destroy(pthread_cond_t *cond);

1. #include <stdio.h> 2. #include <pthread.h> 3. #include "stdlib.h" 4. #include "unistd.h" 5. pthread_mutex_t mutex; 6. pthread_cond_t cond; 7. void hander(void *arg) 8. { 9. free(arg); 10. (void)pthread_mutex_unlock(&mutex); 11. } 12. void *thread1(void *arg) 13. { 14. pthread_cleanup_push(hander, &mutex); 15. while(1) 16. { 17. printf("thread1 is running\n"); 18. pthread_mutex_lock(&mutex); 19. pthread_cond_wait(&cond, &mutex); 20. printf("thread1 applied the condition\n"); 21. pthread_mutex_unlock(&mutex); 22. sleep(4); 23. } 24. pthread_cleanup_pop(0); 25. } 26. void *thread2(void *arg) 27. { 28. while(1) 29. { 30. printf("thread2 is running\n"); 31. pthread_mutex_lock(&mutex); 32. pthread_cond_wait(&cond, &mutex); 33. printf("thread2 applied the condition\n"); 34. pthread_mutex_unlock(&mutex); 35. sleep(1); 36. } 37. } 38. int main() 39. { 40. pthread_t thid1,thid2; 41. printf("condition variable study!\n"); 42. pthread_mutex_init(&mutex, NULL); 43. pthread_cond_init(&cond, NULL); 44. pthread_create(&thid1, NULL, thread1, NULL); 45. pthread_create(&thid2, NULL, thread2, NULL); 46. sleep(1); 47. do 48. { 49. pthread_cond_signal(&cond); 50. }while(1); 51. sleep(20); 52. pthread_exit(0); 53. return 0; 54. }

三、信号量(sem)

1. #include <pthread.h> 2. #include <unistd.h> 3. #include "stdio.h" 4. #include "stdlib.h" 5. static pthread_mutex_t mtx = PTHREAD_MUTEX_INITIALIZER; 6. static pthread_cond_t cond = PTHREAD_COND_INITIALIZER; 7. struct node 8. { 9. int n_number; 10. struct node *n_next; 11. }*head = NULL; 12. 13. static void cleanup_handler(void *arg) 14. { 15. printf("Cleanup handler of second thread./n"); 16. free(arg); 17. (void)pthread_mutex_unlock(&mtx); 18. } 19. static void *thread_func(void *arg) 20. { 21. struct node *p = NULL; 22. pthread_cleanup_push(cleanup_handler, p); 23. while (1) 24. { 25. //这个mutex主要是用来保证pthread_cond_wait的并发性 26. pthread_mutex_lock(&mtx); 27. while (head == NULL) 28. { 29. //这个while要特别说明一下，单个pthread_cond_wait功能很完善，为何 30. //这里要有一个while (head == NULL)呢？因为pthread_cond_wait里的线 31. //程可能会被意外唤醒，如果这个时候head != NULL，则不是我们想要的情况。 32. //这个时候，应该让线程继续进入pthread_cond_wait 33. // pthread_cond_wait会先解除之前的pthread_mutex_lock锁定的mtx， 34. //然后阻塞在等待对列里休眠，直到再次被唤醒(大多数情况下是等待的条件成立 35. //而被唤醒，唤醒后，该进程会先锁定先pthread_mutex_lock(&mtx);，再读取资源 36. //用这个流程是比较清楚的 37. pthread_cond_wait(&cond, &mtx); 38. p = head; 39. head = head->n_next; 40. printf("Got %d from front of queue/n", p->n_number); 41. free(p); 42. } 43. pthread_mutex_unlock(&mtx); //临界区数据操作完毕，释放互斥锁 44. } 45. pthread_cleanup_pop(0); 46. return 0; 47. } 48. int main(void) 49. { 50. pthread_t tid; 51. int i; 52. struct node *p; 53. //子线程会一直等待资源，类似生产者和消费者，但是这里的消费者可以是多个消费者，而 54. //不仅仅支持普通的单个消费者，这个模型虽然简单，但是很强大 55. pthread_create(&tid, NULL, thread_func, NULL); 56. sleep(1); 57. for (i = 0; i < 10; i++) 58. { 59. p = (struct node*)malloc(sizeof(struct node)); 60. p->n_number = i; 61. pthread_mutex_lock(&mtx); //需要操作head这个临界资源，先加锁， 62. p->n_next = head; 63. head = p; 64. pthread_cond_signal(&cond); 65. pthread_mutex_unlock(&mtx); //解锁 66. sleep(1); 67. } 68. printf("thread 1 wanna end the line.So cancel thread 2./n"); 69. //关于pthread_cancel，有一点额外的说明，它是从外部终止子线程，子线程会在最近的取消点，退出 70. //线程，而在我们的代码里，最近的取消点肯定就是pthread_cond_wait()了。 71. pthread_cancel(tid); 72. pthread_join(tid, NULL); 73. printf("All done -- exiting/n"); 74. return 0; 75. }

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

1. 信号量初始化。
   int sem_init (sem_t *sem , int pshared, unsigned int value);
   这是对由sem指定的信号量进行初始化，设置好它的共享选项(linux 只支持为0，即表示它是当前进程的局部信号量)，然后给它一个初始值VALUE。
2. 等待信号量。给信号量减1，然后等待直到信号量的值大于0。
   int sem_wait(sem_t *sem);
3. 释放信号量。信号量值加1。并通知其他等待线程。
   int sem_post(sem_t *sem);
4. 销毁信号量。我们用完信号量后都它进行清理。归还占有的一切资源。
   int sem_destroy(sem_t *sem);

#include <stdlib.h> #include <stdio.h> #include <unistd.h> #include <pthread.h> #include <semaphore.h> #include <errno.h> #define INTERLOCKEDEXCHANGE(x,y) xchg(x,y) #define INTERLOCKEDEXCHANGEADD(v,p) __sync_fetch_and_add(&v, p) //v的值原子添加P的大小 #define INTERLOCKEDINCREMENT(t) __sync_fetch_and_add(&t, 1) //原子+1 #define INTERLOCKEDDECREMENT(t) __sync_fetch_and_sub(&t, 1) //原子-1int nRuning = 0; int bUpdate = 0;#define return_if_fail_void(p) if((p) == 0){printf ("[%s]:func error!/n", __func__);return;} #define return_if_fail_point(p) if((p) == 0){printf ("[%s]:func error!/n", __func__);return p;} typedef struct _PrivInfo { sem_t s1; sem_t s2; time_t end_time; }PrivInfo; static void info_init(PrivInfo* thiz); static void info_destroy(PrivInfo* thiz); static void* pthread_func_1(void* thiz); static void* pthread_func_2(void* thiz); static void* pthread_func_3(void* thiz); static void* pthread_func_4(void* thiz); int main(int argc, char** argv) { pthread_t pt_1 = 0; pthread_t pt_2 = 0; pthread_t pt_3 = 0; pthread_t pt_4 = 0; int ret = 0; PrivInfo* thiz = NULL; thiz = (PrivInfo*)malloc(sizeof(PrivInfo)); if (thiz == NULL) { printf ("[%s]: Failed to malloc priv.\n", __func__); return -1; } info_init(thiz); ret = pthread_create(&pt_1, NULL, pthread_func_1, (void*)thiz); if (ret != 0) { perror ("pthread_1_create:"); } ret = pthread_create(&pt_2, NULL, pthread_func_2, (void*)thiz); if (ret != 0) { perror ("pthread_2_create:"); } ret = pthread_create(&pt_3, NULL, pthread_func_3, (void*)thiz); if (ret != 0) { perror ("pthread_3_create:"); } ret = pthread_create(&pt_4, NULL, pthread_func_4, (void*)thiz); if (ret != 0) { perror ("pthread_4_create:"); } pthread_join(pt_1, NULL); pthread_join(pt_2, NULL); pthread_join(pt_3, NULL); pthread_join(pt_4, NULL); info_destroy(thiz); return 0; }static void info_init(PrivInfo* thiz) { if (thiz == NULL) { printf ("[%s]:func error!\n", __func__); } thiz->end_time = time(NULL) + 60; sem_init(&thiz->s1, 0, 4); sem_init(&thiz->s2, 0, 1); } static void info_destroy(PrivInfo* thiz) { if (thiz == NULL) { printf ("[%s]:func error!\n", __func__); } sem_destroy(&thiz->s1); sem_destroy(&thiz->s2); free(thiz); thiz = NULL; } static void* pthread_func_1(void* thiz) { static int snCount = 1; if (thiz == NULL) { printf ("[%s]:func error!\n", __func__); } PrivInfo* pi = (PrivInfo*)thiz; while (time(NULL) < pi->end_time) { if (bUpdate > 0) { //printf("pthread1:start bUpdate=[%d]\n", bUpdate); goto reload; } else { INTERLOCKEDINCREMENT(nRuning); //+1 printf("pthread1:start nRuning=[%d]\n", nRuning); INTERLOCKEDINCREMENT(snCount); if (snCount%3 == 0) { INTERLOCKEDINCREMENT(bUpdate); printf("pthread1:收到/reload all bUpdate=[%d]\n", bUpdate); } //sem_wait(&pi->s1); sleep(3); //sem_post(&pi->s1); INTERLOCKEDDECREMENT(nRuning); //-1 printf("pthread------1:end------ nRuning=[%d]\n", nRuning); } reload: { sem_wait(&pi->s2); //printf("pthread1: pthread1 lock\n"); if (nRuning == 0 && bUpdate > 0) { sleep(3); INTERLOCKEDDECREMENT(bUpdate); printf("pthread1:执行/reload all nRuning=[%d]\n", nRuning); } //printf("pthread1: pthread1 unlock\n"); sem_post(&pi->s2); } } return thiz; } static void* pthread_func_2(void* thiz) { static int snCount = 1; if (thiz == NULL) { printf ("[%s]:func error!/n", __func__); } PrivInfo* pi = (PrivInfo*)thiz; while (time(NULL) < pi->end_time) { if (bUpdate > 0) { //printf("pthread1:start bUpdate=[%d]\n", bUpdate); goto reload; } else { INTERLOCKEDINCREMENT(nRuning); //+1 printf("pthread2:start nRuning=[%d]\n", nRuning); INTERLOCKEDINCREMENT(snCount); //if (snCount%3 == 0) //{ // INTERLOCKEDINCREMENT(bUpdate); // printf("pthread2:收到/reload all bUpdate=[%d]\n", bUpdate); //} //sem_wait(&pi->s1); sleep(3); //sem_post(&pi->s1); INTERLOCKEDDECREMENT(nRuning); //-1 printf("pthread------2:end------ nRuning=[%d]\n", nRuning); } reload: { sem_wait(&pi->s2); //printf("pthread2: pthread2 lock\n"); if (nRuning == 0 && bUpdate > 0) { sleep(3); INTERLOCKEDDECREMENT(bUpdate); printf("pthread2:执行/reload all nRuning=[%d]\n", nRuning); } //printf("pthread2: pthread2 unlock\n"); sem_post(&pi->s2); } } return thiz; } static void* pthread_func_3(void* thiz) { static int snCount = 1; if (thiz == NULL) { printf ("[%s]:func error!\n", __func__); } PrivInfo* pi = (PrivInfo*)thiz; while (time(NULL) < pi->end_time) { if (bUpdate > 0) { //printf("pthread1:start bUpdate=[%d]\n", bUpdate); goto reload; } else { INTERLOCKEDINCREMENT(nRuning); //+1 printf("pthread3:start nRuning=[%d]\n", nRuning); INTERLOCKEDINCREMENT(snCount); // if (snCount%3 == 0) // { // INTERLOCKEDINCREMENT(bUpdate); // printf("pthread3:收到/reload all bUpdate=[%d]\n", bUpdate); // } //sem_wait(&pi->s1); sleep(3); //sem_post(&pi->s1); INTERLOCKEDDECREMENT(nRuning); //-1 printf("pthread------3:end------ nRuning=[%d]\n", nRuning); } reload: { sem_wait(&pi->s2); //printf("pthread3: pthread3 lock\n"); if (nRuning == 0 && bUpdate > 0) { sleep(3); INTERLOCKEDDECREMENT(bUpdate); printf("pthread3:执行/reload all nRuning=[%d]\n", nRuning); } //printf("pthread3: pthread1 unlock\n"); sem_post(&pi->s2); } } return thiz; } static void* pthread_func_4(void* thiz) { static int snCount = 1; if (thiz == NULL) { printf ("[%s]:func error!/n", __func__); } PrivInfo* pi = (PrivInfo*)thiz; while (time(NULL) < pi->end_time) { if (bUpdate > 0) { //printf("pthread1:start bUpdate=[%d]\n", bUpdate); goto reload; } else { INTERLOCKEDINCREMENT(nRuning); //+1 printf("pthread4:start nRuning=[%d]\n", nRuning); INTERLOCKEDINCREMENT(snCount); // if (snCount%3 == 0) // { // INTERLOCKEDINCREMENT(bUpdate); // printf("pthread4:收到/reload all bUpdate=[%d]\n", bUpdate); // } //sem_wait(&pi->s1); sleep(3); //sem_post(&pi->s1); INTERLOCKEDDECREMENT(nRuning); //-1 printf("pthread------4:end------ nRuning=[%d]\n", nRuning); } reload: { sem_wait(&pi->s2); //printf("pthread4: pthread4 lock\n"); if (nRuning == 0 && bUpdate > 0) { sleep(3); INTERLOCKEDDECREMENT(bUpdate); printf("pthread4:执行/reload all nRuning=[%d]\n", nRuning); } //printf("pthread4: pthread4 unlock\n"); sem_post(&pi->s2); } } return thiz; } /* int sem_wait(sem_t*sem);intsem_trywait(sem_t *sem); intsem_timedwait(sem_t *sem, const struct timespec *abs_timeout); struct timespec */