linux系統7--線程同步

線程同步

• 1.線程同步的概念
• 2.鎖
• • 2.1互斥鎖(互斥量)
  • 2.2 死鎖
  • 2.3 讀寫鎖
  • 2.4 條件變量
  • 2.5 信号量
  • 2.6 自旋鎖
• 3.原子操作：

1.線程同步的概念

所謂線程同步，就是多個線程同時通路同一資源，多個線程協同步調，先後處理某件事情。那麼如何實作線程同步呢？就是利用"鎖"來實作的。下圖為實作線程同步的基本操作：

如上圖，線程在通路資料前必須加鎖，加鎖的目的是為了對線程進行阻塞，若鎖已經被另外的線程鎖上了，那麼目前線程進入阻塞态，直到鎖被打開，解除阻塞。若沒有上鎖，目前進行直接加鎖。

上圖中，資源被線程1直接上鎖，那麼線程2就進入阻塞狀态。直到線程1解鎖，然後線程2解除阻塞，加鎖。如此往複。

2.鎖

綜上所述，我們要使用線程同步，就必須使用鎖，那麼鎖怎麼加又是怎麼打開呢？鎖的種類呢？下面将一一介紹：

2.1互斥鎖(互斥量)

互斥鎖：pthread_mutex_t mutext;

互斥鎖提供一個可以在同一時間，隻讓一個線程通路臨界資源的的操作接口。互斥鎖讓多個線程通路共享資料的時候是串行的。互斥鎖的值，要麼是0要麼是1，1表示解鎖，0表示加鎖

互斥鎖的使用步驟：

• 建立互斥鎖：pthread_mutex_t mutex;
• 初始化這把鎖：pthread_mutex_init(&mutex,NULL)

• 如何對資源加鎖：在操作共享資源的代碼之前加鎖，之後解鎖

  使用示例：

  pthread_mutex_lock(&mutex);

  //共享資源代碼//

  pthread_mutex_unlock(&mutex);

互斥鎖的相關函數

<1> 初始化互斥鎖

pthread_mutex_init(

pthread_mutex_t *restrict mutex, //用其他的指針指向此位址是不管用的

const pthread_mutexattr_t *restrict attr//一般為NULL

);

<2> 銷毀互斥鎖

pthread_mutex_destroy(pthread_mutex_t *mutex);

<3> 加鎖

pthread_mutex_lock(pthread_mutex_t *mutex);

-參數：mutex：

沒有被上鎖：目前線程會将這把鎖鎖上

被鎖上了：目前線程阻塞

鎖被打開之後：線程解除阻塞

<4>嘗試加鎖, 失敗傳回, 不阻塞

pthread_mutex_trylock(pthread_mutex_t *mutex);

• 沒有鎖上：目前線程會給這把鎖加鎖
• 如果鎖上了：不會阻塞，傳回

• 使用示例：

  if(pthread_mutex_trylock(&mutex)==0）

  {

  // 則說明嘗試加鎖，并且成功了

  // 接下來進行通路共享資源操作

  }

  else

  {

  //表示 加鎖失敗

  //則進行錯誤處理

  // 或者 等一會，再次嘗試加鎖

  }

  <6> 解鎖

  pthread_mutex_unlock(pthread_mutex_t *mutex);

2.2 死鎖

死鎖不是一種鎖，而是一種現象。

造成死鎖的原因1：自己鎖自己，如下代碼：

pthread_mutex_lock(&mutex); 
pthread_mutex_lock(&mutex); 
//資源代碼塊
pthread_mutex_unlock(&mutex); 
           

顯然上面代碼使用同意吧鎖，鎖了兩次，但是隻解鎖了一次。說明該資源依舊是被鎖的。這是由于不規範的程式設計，忘了加鎖後進行解鎖。是以造成了死鎖。

解決方法：加鎖後一定要解鎖

** 造成死鎖的原因2：兩個線程都被阻塞了**

如圖：

上圖所示：

首先：

線程1對共享資源A進行加鎖：A鎖

線程2對共享資源B進行加鎖：B鎖

接着：

線程1通路共享資源B，對B加鎖。此時由于線程2對B加鎖了，是以線程1被阻塞在B鎖上，加鎖失敗。

線程2通路共享資源A，對A加鎖。此時由于線程1對A加鎖了，是以線程2被阻塞在A鎖上，加鎖失敗。

這樣就導緻兩個線程都被分别阻塞在AB鎖上了。那麼如何解決這件事呢？

• 方法1：讓線程按照一定的順序去通路共享資源。
• 方法2：在通路其他鎖的時候，需要先将自己的鎖打開。

2.3 讀寫鎖

讀寫鎖：pthread_rwlock_t lock

讀寫鎖将對資源代碼的通路分為讀寫兩種模式，這大大的提高了并發效率。讀寫鎖相較于互斥鎖具有更好的性能，因為假如以讀模式加鎖後，當有多個線程對資源以讀模式加鎖時，并不會造成這些線程阻塞在等待解鎖。

讀寫鎖隻有一把鎖，不過是将資源的分文分為了讀寫兩種模式。

讀寫鎖類型：

• 讀鎖：對記憶體讀操作
• 寫鎖：對記憶體寫操作

讀寫鎖特性

<1>線程A加讀鎖成功，此時又來了三個線程想要進行讀操作，那麼這三個線程可以加鎖成功

• 讀共享—讀是并行處理的

<2> 線程A加寫鎖成功，此時又來了三個線程想要進行讀操作，那麼這三個線程被阻塞

• 寫獨占

<3> 線程A加讀鎖成功，此時又來了線程B想要進行寫操作，那麼線程B被阻塞。此時又來了線程C想要進行讀操作，那麼線程C被阻塞

• 讀寫不能同時進行
• 寫的優先級更高

根據讀寫鎖特性的一些場景練習

<1> 線程A持有寫鎖，線程B請求加讀鎖

• 線程B被阻塞：寫鎖優先級更高

<2> 線程A持有讀鎖，線程B請求加寫鎖

• 線程B阻塞：讀寫不能同時進行

<3> 線程A擁有讀鎖，線程B請求讀鎖

• 線程B加鎖成功：讀共享

<4> 線程A持有讀鎖，然後線程B請求寫鎖，然後線程C請求讀鎖

• 線程B被阻塞，線程C被阻塞：讀寫不能同時進行，寫的優先級高
• 若線程A解鎖，線程B加寫鎖成功，C繼續阻塞
• B解鎖，C加讀鎖成功

<5> 線程A持有寫鎖，然後線程B請求讀鎖，然後線程C請求寫鎖

• 線程B被阻塞，線程C被阻塞
• 若線程A解鎖，線程C加寫鎖成功，B繼續阻塞：寫的優先級更高
• C解鎖，B加讀鎖成功

讀寫鎖的使用場景

• 互斥鎖的使用場景為：讀寫串行時

• 讀寫鎖：

  由讀寫鎖的讀寫特性，我們知道，讀是并行的，寫是串行的。那麼讀寫鎖的使用場景應該是程式中的讀操作的數量大于寫操作的數量時。

讀寫鎖的相關函數

<1> 初始化讀寫鎖

pthread_rwlock_init(

pthread_rwlock_t *restrict rwlock,

const pthread_rwlockattr_t *restrict attr

);

<2> 銷毀讀寫鎖

pthread_rwlock_destroy(pthread_rwlock_t *rwlock);

<3>加讀鎖

pthread_rwlock_rdlock(pthread_rwlock_t *rwlock);

• 上一次加寫鎖，但是還沒解鎖，則阻塞

<4> 嘗試加讀鎖

pthread_rwlock_tryrdlock(pthread_rwlock_t *rwlock);

• 傳回值：

  加鎖成功：0

  失敗：錯誤号

<5> 加寫鎖

pthread_rwlock_wrlock(pthread_rwlock_t *rwlock);

• 上一次加寫鎖，但是還沒有解鎖，則阻塞
• 上一次加讀鎖，但是還沒有解鎖，則阻塞

<6>嘗試加寫鎖

pthread_rwlock_trywrlock(pthread_rwlock_t *rwlock);

<7>解鎖

pthread_rwlock_unlock(pthread_rwlock_t *rwlock);

讀寫鎖使用示例：

對同一資源，使用5次讀線程，3次寫線程

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>
//定義全局變量共享資源
int num = 0;
//定義讀寫鎖
pthread_rwlock_t lock;
//read_thread函數的實作
void* read_thread(void *arg)
{
    while(1)
    {
        //加讀鎖
        pthread_rwlock_rdlock(&lock);
        printf("read number = %d\n",num);
        //解讀鎖
        pthread_rwlock_unlock(&lock);
        usleep(500);
    }
    return NULL;
}
//write_thread函數的實作
void *write_thread(void *arg)
{
    while(1)
    {
        //加寫鎖
        pthread_rwlock_wrlock(&lock);
        num++;
        printf("wrirte number = %d\n",num);
        //解鎖
        pthread_rwlock_unlock(&lock);
        usleep(500);
    }
    return NULL;
}
int main(void)
{
    //建立八個子線程
    pthread_t thread[8];
    //初始化讀寫鎖
    pthread_rwlock_init(&lock,NULL);
    //建立5個讀線程
    for(int i = 0;i < 5;++i)
    {
       int ret =  pthread_create(&thread[i],NULL,read_thread,NULL);
       if(ret != 0)
       {
           printf("第 %d 個線程建立失敗\n",i);
           printf("error:%s\n",strerror(ret));
       }
    }
    //建立3個寫線程
    for(int i = 5;i < 8;++i)
    {
        int ret = pthread_create(&thread[i],NULL,write_thread,NULL);
        if(ret != 0)
        {
            printf("第%d個線程建立失敗\n",i);
            printf("error:%s\n",strerror(ret));
        }
    }
    //線程回收
    for(int i = 0;i < 8;++i)
    {
        pthread_join(thread[i],NULL);
    }
    //讀寫鎖的銷毀
    pthread_rwlock_destroy(&lock);
    return 0;
}
           

運作結果：

顯然，資料是按照升序逐個增加的

2.4 條件變量

條件變量：pthread_cond_t cond;

條件變量不是鎖，但是能是夠實作阻塞線程的功能。條件變量一般與互斥鎖聯合使用。當條件不滿足時，條件變量阻塞線程；當條件滿足時，會通知相應被阻塞的一個或多個線程開始工作。

一個例子：生産者-消費者模型

所謂生産者-消費者模型，就是生産者生産"資料"，消費者消費“資料”。如生産者線程用于給連結清單中插入結點，消費者線程列印尾結點資料，并删除尾結點。

若我們直接使用互斥鎖，那麼生産者和多個消費者之間，及消費者之間都會去“搶”鎖。但是，若生産者沒有搶到，那麼就沒有“資料"讓消費者消費，那麼消費者之間“搶”鎖就是做無用功，因為就算他們搶到了，也沒有進行消費。這就造成了資源浪費。

但是我們使用了條件變量，我們就可以使用條件變量讓消費者線程阻塞，讓生産者通知被阻塞的線程開始工作。這樣的話，就不會出現上面資源浪費的情況。

條件變量相關函數：

<1> 初始化一個條件變量

pthread_cond_init(

pthread_cond_t *restrict cond,

const pthread_condattr_t *restrict attr);

<2> 銷毀一個條件變量

pthread_cond_destroy(pthread_cond_t *cond);

<3> 阻塞等待一個條件變量

pthread_cond_wait(pthread_cond_t *restrict cond,

pthread_mutex_t *restrict mutex);

• 函數作用：阻塞線程

  若該函數阻塞線程，則将已經上鎖的mutex解鎖

  若該函數解除阻塞，會對互斥鎖加鎖

<4> 限時等待一個條件變量-----------阻塞一定的時長

pthread_cond_timedwait(

pthread_cond_t *restrict cond,

pthread_mutex_t *restrict mutex,

const struct timespec *restrict abstime//阻塞時長);

<5> 喚醒至少一個阻塞在條件變量上的線程

pthread_cond_signal(pthread_cond_t *cond);

<6> 喚醒全部阻塞在條件變量上的線程

pthread_cond_broadcast(pthread_cond_t *cond);

生産者-消費者模型實作

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>
//建立互斥鎖
pthread_mutex_t mutex;
//建立條件變量
pthread_cond_t cond;
//定義生産出的結點
typedef struct node
{
    int data;
    struct node *next;
}NODE;
//定義指向連結清單頭的頭指針
NODE *header = NULL;
//生産者線程
void *producer(void *arg)
{
    while(1)
    {
        //建立節點
        NODE *node = (NODE*)malloc(sizeof(NODE));
        //設定節點資料
        node->data = (rand() % 1000);//0-1000的随機數
        //加鎖
        pthread_mutex_lock(&mutex);
        //使用頭插法将結點插傳入連結表
        node ->next = header;
        header = node;
        //列印生産出來的結點資料
        printf("producer data = %d\n",node ->data);
        //解鎖
        pthread_mutex_unlock(&mutex);
        //通知消費者鎖的狀态為解除阻塞，
        pthread_cond_signal(&cond);//若生産者沒有生産結點，說明這句話執行并沒有執行,那麼消費者鎖的狀态為阻塞
         sleep(rand()%3);//睡1-3s
    }
    return NULL;
}
//消費者線程
void *customer(void *arg)
{
    while(1)
    {
        //删除結點
        if(header == NULL)//首先判斷結點是否為空
        {
            //若為空則執行以下操作
            //等待阻塞
            pthread_cond_wait(&cond,&mutex);//若結點為空，說明生産者那邊沒有解除阻塞狀态，此時該函數阻塞線程。互斥鎖為解鎖狀态。直到生産者那麼有了資料，那麼就會解除阻塞，那麼互斥鎖就會上鎖，保證其他消費者不會通路到下面代碼資料
        }
        //結點非空
        //頭删法删除結點
        //定義代删除結點指針
        NODE *del_node = header;
        //重建結點關系
        header = del_node ->next;
        //列印删除結點的資料
        printf("customer data = %d\n",del_node ->data);
        //釋放已删除結點的資源
        free(del_node);
        //解鎖
        pthread_mutex_unlock(&mutex);
        sleep(rand()%3);//睡1-3s
    }
    return NULL;
}
int main(void)
{
    //建立兩個子線程
    pthread_t thread_producer;
    pthread_t thread_cstm;
    //初始化互斥鎖
    pthread_mutex_init(&mutex,NULL);
    //初始化條件變量
    pthread_cond_init(&cond,NULL);
    //建立生産者線程
    pthread_create(&thread_producer,NULL,producer,NULL);
    //建立消費者線程
    pthread_create(&thread_cstm,NULL,customer,NULL);
    //線程資源回收
    pthread_join(thread_producer,NULL);
    pthread_join(thread_cstm,NULL);
    //銷毀互斥鎖
    pthread_mutex_destroy(&mutex);
    //銷毀條件變量
    pthread_cond_destroy(&cond);
    return 0;
}
           

2.5 信号量

信号量：sem_t sem

信号量就是加強版的互斥鎖，使用互斥鎖對多線程加鎖，使同一時間隻允許一個線程對資源進行通路，具有唯一性和排他性。但是互斥鎖無法限制對資源的通路順序，但是使用信号量，它通過兩個原子操作來通路資源，實作了對資源的通路是有序的。（有關原子操作，下一節介紹）

互斥鎖的值隻能為0或1，但是信号量可以為負數，0表示解鎖，非零加鎖。

互斥鎖的加鎖和解鎖隻能由同一個線程中，但是信号量可以由一個線程加鎖另一個線程解鎖

信号量的相關函數

<1> 初始化信号量：sem_init(sem_t *sem, int pshared, unsigned int value);

• pshared

  0 - 線程同步

  1 - 程序同步

• value

  最多有幾個線程操作共享資料

<2> 銷毀信号量：sem_destroy(sem_t *sem);

<3> 加鎖 sem–：對sem–會加鎖

sem_wait(sem_t *sem);

• 調用一次相當于對sem做了–操作
• 如果sem值為0, 線程會阻塞

<4>嘗試加鎖 ：sem_trywait(sem_t *sem);

• sem == 0, 加鎖失敗, 不阻塞, 直接傳回

<5> 限時嘗試加鎖：sem_timedwait(sem_t *sem, xxxxx);

<6> 解鎖 sem++：對sem++會解鎖

sem_post(sem_t *sem);

• 對sem做了++操作

使用信号量實作生産者-消費者模型

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <unistd.h>
#include <semaphore.h>
//定義連結清單結點
typedef struct node
{
    int data;
    struct node * next;
}NODE;
//定義指向連結清單頭的指針
NODE *header = NULL;
//定義信号燈
sem_t sem_producer;
sem_t sem_custmoer;
void *producer_thread(void*arg)
{
    //生産者生産連結清單結點
    while(1)
    {
        //使用信号量加鎖,使sen--
        sem_wait(&sem_producer);//隻有sem不為0時，線程才會解除阻塞，解除阻塞後又對sem--，sem加鎖，保證沒有資料被産生時，消費者不能消費資料。隻有一下代碼執行後，産生了資料，同時sem解鎖了，消費者才能消費資源。
        //給建立的結點開辟空間
        NODE *node = (NODE*)malloc(sizeof(NODE));
        //給結點資料指派
        node->data = rand()%1000;//1000以内的随機整數
        //使用頭插法将結點插傳入連結表
        node->next = header;
        header = node;
        //列印結點資料
        printf("product data = %d\n",node ->data);
        //解鎖,使其sem++
        sem_post(&sem_custmoer);  
        sleep(rand()%3);
    }
    return NULL;
}
void *customer_thread(void*arg)
{
    while(1)
    {
        //使用信号量進行消費者判斷，判斷結果将決定該函數是否阻塞
        sem_wait(&sem_custmoer);//隻有sem不為0時，線程才會解除阻塞，解除阻塞後又對sem--，加鎖，使其他消費者線程不能通路下面資源
        //解除阻塞後的操作
        //删除頭結點資料        
        //首先儲存頭結點指針
        NODE *del_node = header;
        //重建結點關系
        header = del_node->next;
        //列印删除結點的資料
        printf("custome data = %d\n",del_node ->data);
        //釋放删除結點的記憶體
        free(del_node);
        //解鎖使其sem++
        sem_post(&sem_producer);
        sleep(rand()%3);
    }
    return NULL;
}
int main(void)
{
    //建立兩個線程
    pthread_ t thread_p;
    pthread_t  thread_c;
    //初始化信号量
    sem_init(&sem_producer,0,4);
    sem_init(&sem_custmoer,0,0);
    //chuangjianxiancheng
    pthread_create(&thread_p,NULL,producer_thread,NULL);//生産者線程
    pthread_create(&thread_c,NULL,customer_thread,NULL);//消費者線程
    //線程資源的回收
    pthread_join(thread_p,NULL);
    pthread_join(thread_c,NULL);
    //銷毀信号量
    sem_destroy(&sem_producer);
    sem_destroy(&sem_custmoer);
    return 0;
}
           

2.6 自旋鎖

參考：https://www.cnblogs.com/cxuanBlog/p/11679883.html

3.原子操作：

cpu處理一個指令時，線程/程序在處理完這個指令之前不會丢失cpu的。如下：

非原子操作：

void* producer(void* arg)
  {
      // 一直生産
      while(1)
      {   
          // 建立一個連結清單的節點
          Node* newNode = (Node*)malloc(sizeof(Node));
          // init
          newNode->data = rand() % 1000;
          //****************************************//
          newNode->next = head;
          head = newNode;
          printf("+++ producer: %d\n", head->data);
          //****************************************//
          sleep(rand()%3);
      }   
      return NULL;
  } 
           

上面這段代碼，中間被括起來的代碼，由于head被放于全局區，那麼執行這段代碼時，可能會失去cpu，那麼結果就會出錯。

我們可以使用鎖來模拟原子操作

void* producer(void* arg)
  {
      // 一直生産
      while(1)
      {   
          // 建立一個連結清單的節點
          Node* newNode = (Node*)malloc(sizeof(Node));
          // init
          newNode->data = rand() % 1000;
          pthread_mutex_lock(&mutex);
          newNode->next = head;
          head = newNode;
          printf("+++ producer: %d\n", head->data);
          pthread_mutex_unlock(&mutex);
          sleep(rand()%3);
      }   
      return NULL;
  } 
           

使用互斥鎖，不可能讓線程失去cpu的。故結果是對的。