Linux系统编程---线程同步-易微帮

一、同步概念

同步即协同步调，按预定的先后次序运行。

协同步调，对公共区域数据【按序】访问，防止数据混乱，产生与时间有关的错误。

数据混乱的原因：

资源共享(独享资源则不会)

调度随机(意味着数据访问会出现竞争)

线程间缺乏必要同步机制

二、锁

1. 互斥锁

linux中提供一把互斥锁mutex（也称之为互斥量）。

建议锁！对公共数据进行保护。所有线程【应该】在访问公共数据前先拿锁再访问。但锁本身不具备强制性。

数据共享导致的混乱pthrd_shared.c：

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

//子线程
void *tfn(void *arg)
{
        srand(time(NULL));

        while(1)
        {
                printf("hello ");
                sleep(rand() % 3); //模拟长时间操作共享资源，导致cpu易主，产生与时间有关的错误
                printf("world\n");
                sleep(rand() % 3);
        }
        return NULL;
}

int main(void)
{
        pthread_t tid;
        srand(time(NULL));

        // 创建线程
        pthread_create(&tid,NULL,tfn,NULL);
        while(1)
        {
                printf("HELLO ");
                sleep(rand() % 3);
                printf("WORLD\n");
                sleep(rand() % 3);
        }

        //回收线程
        pthread_join(tid,NULL);

        return 0;
}

输出为：

HELLO hello WORLD
world
hello HELLO WORLD
world
HELLO hello WORLD
world
hello HELLO world
hello WORLD
world
HELLO hello world

pthread_mutex_函数

初始化：int pthread_mutex_init(pthread_mutex_t *restrict mutex,const pthread_mutexattr_t *restrict attr);

销毁：int pthread_mutex_destroylock(pthread_mutex_t *mutex);

上锁：int pthread_mutex_lock(pthread_mutex_t *mutex);

try锁：int pthread_mutex_trylock(pthread_mutex_t *mutex);

解锁：int pthread_mutex_unlock(pthread_mutex_t *mutex);

以上函数的返回值都是：成功返回0，失败返回错误号；

restrict关键字

用来限定指针变量。被该关键字限定的指针变量所指向的内存操作，必须由本指针完成。

使用mutex(互斥量、互斥锁)一般步骤:

pthread_mutex_t lock；创建锁

pthread_mutex_init；初始化

pthread_mutex_lock；加锁

访问共享数据

pthread_mutex_unlock；解锁

pthread_mutex_destroy；销毁锁

初始化互斥量：

pthread_mutex_t mutex;

1. 动态初始化：pthread_mutex_init(&mutex, NULL);

2. 静态初始化：pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

注意事项：

1. 尽量保证锁的粒度，越小越好。(访间共享数据前，加锁。访问结束【立即】解锁。)
2. 互斥锁：本质是结构体。我们可以看成整数。初值为1。(pthread_mutex_init(函数调用成功。))

3. 加锁：--操作，阻塞线程。
4. 解锁：++操作，换醒阻塞在锁上的线程。
5. try锁：尝试加锁，成功--。失败,返回。同时设置错误号EBUSY

修改上面pthrd_shared.c的代码，使用锁实现互斥访问共享区：

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

pthread_mutex_t mutex;//定义一把互斥锁

//子线程
void *tfn(void *arg)
{
        srand(time(NULL));

        while(1)
        {
                pthread_mutex_lock(&mutex);//加锁
                printf("hello ");
                sleep(rand() % 3);//模拟长时间操作共享资源，导致cpu易主，产生与时间有关的错误
                printf("world\n");
                pthread_mutex_unlock(&mutex);//解锁
        
                sleep(rand() % 3);
        }
        return NULL;
}
         
int main(void)
{
        pthread_t tid;
        srand(time(NULL));
        int ret = pthread_mutex_init(&mutex,NULL);//初始化互斥锁
        if (ret != 0)
        {
                fprintf(stderr,"mutex init error:%s\n",strerror(ret));
                exit(1);        
        }

        // 创建线程
        pthread_create(&tid,NULL,tfn,NULL);
        while(1)
        {
                pthread_mutex_lock(&mutex);//加锁
                printf("HELLO ");
                sleep(rand() % 3);
                printf("WORLD\n");
                pthread_mutex_unlock(&mutex);//解锁
                sleep(rand() % 3);
        }

        //回收线程
        pthread_join(tid,NULL);

        pthread_mutex_destroy(&mutex);//销毁互斥锁
        return 0;
}

输出为：

HELLO WORLD
hello world
HELLO WORLD
hello world
HELLO WORLD
hello world

2. 死锁：对锁使用不恰当导致的现象

1. 对一个锁反复lock。

2. 两个线程，各自持有一把锁，请求另一把。

情况 1：对一个锁反复 lock

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

pthread_mutex_t mutex;

void *lockTwice(void *arg) 
{
    pthread_mutex_lock(&mutex); // 第一次获取锁
    printf("Lock acquired once.\n");
    pthread_mutex_lock(&mutex); // 尝试再次获取同一个锁，导致死锁
    printf("Lock acquired twice.\n");
    pthread_mutex_unlock(&mutex);
    pthread_mutex_unlock(&mutex);

    return NULL;
}

int main() 
{
    pthread_t tid;
    pthread_mutex_init(&mutex, NULL);
    
    pthread_create(&tid, NULL, lockTwice, NULL);
    pthread_join(tid, NULL);

    pthread_mutex_destroy(&mutex);
    return 0;
}

在这个示例中，我们尝试在同一线程中两次锁定同一个互斥锁。因为 pthread_mutex_t 默认是非递归的，第二次尝试锁定会导致线程阻塞，从而产生死锁。

情况 2：两个线程，各自持有一把锁，请求另一把

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

pthread_mutex_t lock1, lock2;

void *thread1Func(void *arg) 
{
    pthread_mutex_lock(&lock1);
    printf("Thread 1 acquired lock 1\n");
    sleep(1); // 增加死锁发生的可能性
    pthread_mutex_lock(&lock2);
    printf("Thread 1 acquired lock 2\n");
    pthread_mutex_unlock(&lock2);
    pthread_mutex_unlock(&lock1);
    return NULL;
}

void *thread2Func(void *arg) {
    pthread_mutex_lock(&lock2);
    printf("Thread 2 acquired lock 2\n");
    sleep(1); // 增加死锁发生的可能性
    pthread_mutex_lock(&lock1);
    printf("Thread 2 acquired lock 1\n");
    pthread_mutex_unlock(&lock1);
    pthread_mutex_unlock(&lock2);
    return NULL;
}

int main() 
{
    pthread_t t1, t2;
    pthread_mutex_init(&lock1, NULL);
    pthread_mutex_init(&lock2, NULL);
    
    pthread_create(&t1, NULL, thread1Func, NULL);
    pthread_create(&t2, NULL, thread2Func, NULL);

    pthread_join(t1, NULL);
    pthread_join(t2, NULL);

    pthread_mutex_destroy(&lock1);
    pthread_mutex_destroy(&lock2);
    return 0;
}

这个示例中，thread1Func 先锁定 lock1，然后尝试锁定 lock2。同时，thread2Func 先锁定 lock2，然后尝试锁定 lock1。这种交叉锁定容易导致死锁，因为每个线程都在等待对方释放另一把锁。

3. 读写锁rwlock：

pthread_rwlock_函数

pthread_rwlock_t rwlock;        用于定义一个读写锁变量

pthread_rwlock_init(&rwlock, NULL);

pthread_rwlock_rdlock(&rwlock);         tryrdlock

pthread_rwlock_wrlock(&rwlock);         trywrlock

pthread_rwlock_unlock(&rwlock);

pthread_rwlock_destroy(&rwlock);

参数与互斥锁类似

以上函数的返回值都是：成功返回0，失败返回错误号

与互斥量类似，但读写锁允许更高的并行性。其特性为：写独占，读共享

锁只有一把。以读方式给数据加锁------读锁。以写方式给数据加锁------写锁。

读共享，写独占。

写锁优先级高。

相较于互斥量而言，当读线程多的时候，提高访问效率

读写锁是“读模式加锁”时，既有试图以写模式加锁的线程，也有试图以读模式加锁的线程。那么读写锁会阻塞随后的读模式锁请求。优先满足写模式锁。读锁、写锁并行阻塞，写锁优先级高

代码示例验证写锁优先级高：

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

int counter;
pthread_rwlock_t rwlock;//全局的读写锁

//3个线程不定时写统一全局资源，5个线程不定时读统一全局资源
void *th_write(void *arg)
{
        int t;
        int i = (int)arg;
        while (1) 
        {
                pthread_rwlock_wrlock(&rwlock);//以写模式加锁，写独占
                t = counter;
                usleep(1000);
                printf("===write %d: %lu: counter=%d ++counter=%d\n",i,pthread_self(),t,++counter);
                pthread_rwlock_unlock(&rwlock);
                usleep(10000);
        }
        return NULL;
}

void *th_read(void *arg)
{
        int i = (int)arg;
        while (1) 
        {
                pthread_rwlock_rdlock(&rwlock);//读锁共享
                printf("-----read %d: %lu: %d\n",i,pthread_self(),counter);
                pthread_rwlock_unlock(&rwlock);
                usleep(2000);
        }
        return NULL;
}

int main(void)
{
        int i;
        pthread_t tid[8];

        pthread_rwlock_init(&rwlock,NULL);

        for (i = 0;i < 3;i++)
                pthread_create(&tid[i],NULL,th_write,(void*)i);

        for (i = 0; i < 5; i++)
                pthread_create(&tid[i+3],NULL,th_read,(void*)i);
        
        for (i = 0;i< 8; i++)
                pthread_join(tid[i],NULL);

        pthread_rwlock_destroy(&rwlock);

        return 0;
}

三、条件变量

条件变量本身不是锁！但它也可以造成线程阻塞，通常与互斥锁配合使用。

主要应用函数：

pthread_cond_init函数初始化一个条件变量

pthread_cond_destroy函数销毁一个条件变量

pthread_cond_wait函数阻塞等待一个条件变量

pthread_cond_timedwait函数限时等待一个条件变量

pthread_cond_signal函数唤醒至少一个阻塞在条件变量上的线程

pthread_cond_broadcast函数唤醒全部阻塞在条件变量上的线程

参数与互斥锁类似

以上6 个函数的返回值都是：成功返回0，失败直接返回错误号。

pthread_cond_t类型用于定义条件变量

pthread_cond_t cond;

pthread_cond_wait函数

pthread_cond_wait(&cond, &mutex);

阻塞等待一个条件变量

作用：

阻塞等待条件变量cond（参1）满足

解锁已经掌握的互斥锁【解锁互斥量】（相当于 pthread_mutex_unlock(&mutex)），1 2两步为一个原子操作(一起执行，不可分割)

当被唤醒，pthread_cond_wait函数返回时，解除阻塞并重新申请获取互斥锁，即重新加锁（相当于pthread_mutex_lock(&mutex);）

1. 当调用pthread_cond_wait函数时，它在阻塞等待条件满足时，会解锁

2. 当条件满足后，重新加锁

3. pthread_cond_signal()、pthread_cond_broadcast()函数会发送条件满足的信号

4. 通过pthread_cond_timewait()函数来限时等待一个条件变量

条件变量的生产者消费者模型分析：

条件变量实现生产者消费者代码：

#include<stdio.h>
#include<stdlib.h>
#include<string.h>
#include<unistd.h>
#include<errno.h>
#include<pthread.h>

void err_thread(int ret,char *str)
{
        if (ret != 0)
        {
                fprintf(stderr,"%s:%d\n",str,strerror(str));
                pthread_exit(NULL);
        }
}

//创建共享数据
struct msg
{
        int num;
        struct msg *next;
};

struct mas *head;

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;//定义并且初始化一个互斥量
pthread_cond_t has_data = PTHREAD_COND_INITIALIZER;//定义并且初始化一个条件变量

void *producer(void *arg)
{
        while(1){       //添加循环多次生产
        struct msg *mp = malloc(sizeof(struct msg));

        mp->num = rand() % 1000 + 1;//模拟生产一个数据
        printf("---produce %d\n",mp->num);

        pthread_mutex_lock(&mutex);//加锁互斥量
        mp->next = head;//头插法插入链表,写公共区域
        head = mp;
        pthread_mutex_unlock(&mutex); //解锁互斥量

        pthread_cond_signal(&has_data); //唤醒在阻塞条件变量has_data上的线程

        sleep(rand() % 3);
        }
        return NULL;
}

void *consumer(void *arg)
{
        while(1){
        struct msg *mp;
        pthread_mutex_lock(&mutex);//加锁互斥量
        if (head == NULL)
        {
                pthread_cond_wait(&has_data,&mutex);//阻塞等待条件变量，解锁。pthread_connd_wait返回时，重新加锁mutex
        }

        mp = head;
        head = mp->next;

        pthread_mutex_unlock(&mutex);//解锁互斥量
        printf("------consumer:%d\n",mp->num);

        free(mp);
        sleep(rand() % 3);
        }
        return NULL;
}

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

        srand(time(NULL));

        ret = pthread_create(&pid,NULL,producer,NULL);
        if (ret != 0)
        {
                err_thread(ret,"pthread_create producer error");//生产者
        }
        ret = pthread_create(&cid,NULL,consumer,NULL);
        if (ret != 0)
        {
                err_thread(ret,"pthread_create consumer error");//消费者
        }

        pthread_join(pid,NULL);
        pthread_join(cid,NULL);

        return 0;
}

输出为：

---produce 330
------consumer id:139864563013376:330
---produce 839
------consumer id:139864554620672:839
---produce 126
------consumer id:139864554620672:126

多个消费者使用while做：

if (ret != 0)
        {
                err_thread(ret,"pthread_create consumer error");//消费者
        }

复制多份创建消费者的代码，并且在consumer回调函数中将if (head == NULL)修改为while (head == NULL)。若未进行修改，则出现段错误

输出为：

---produce 783
---produce 284
---produce 115
---produce 203
------consumer id:139795136452352:203
------consumer id:139795144845056:115
---produce 199
------consumer id:139795153237760:199
------consumer id:139795153237760:284
------consumer id:139795136452352:783

条件变量signal注意事项：

pthread_cond_signal(): 唤醒阻塞在条件变量上的 (至少)一个线程。

pthread_cond_broadcast()：唤醒阻塞在条件变量上的所有线程。

四、信号量

应用于线程、进程间同步。

相当于初始化值为 N 的互斥量。 N值，表示可以同时访问共享数据区的线程数

sem_函数：

sem_t sem; 定义类型,本质是结构体

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

参数：

        sem：信号量

        pshared： 0：用于线程间同步

                        1：用于进程间同步

        value：N值。（指定同时访问的线程数）

sem_destroy();

sem_wait();一次调用，做一次-- 操作，当信号量的值为 0 时，再次 -- 就会阻塞。

（对比pthread_mutex_lock）

sem_trywait();

sem_timedwait();

sem_post();一次调用，做一次++ 操作，当信号量的值为 N 时, 再次 ++ 就会阻塞。

（对比 pthread_mutex_unlock）

以上6 个函数的返回值都是：成功返回0，失败返回-1，同时设置errno。

信号量的初值，决定了占用信号量的线程的个数

信号量实现的生产者消费者：

分析：

代码示例：

 /*信号量实现 生产者 消费者问题*/
#include<stdio.h>  
#include<stdlib.h>  
#include<unistd.h>  
#include<string.h>  
#include<pthread.h>  

#define NUM 5                 
   
int queue[NUM];                                     //全局数组实现环形队列  
sem_t blank_number, product_number;                 //空格子信号量, 产品信号量  
   
void *producer(void *arg)  
{  
     int i = 0;  
     while (1) 
    {  
         sem_wait(&blank_number);                    //生产者将空格子数--,为0则阻塞等待  
         queue[i] = rand() % 1000 + 1;               //生产一个产品  
         printf("----Produce---%d\n", queue[i]);          
         sem_post(&product_number);                  //将产品数++  
   
         i = (i+1) % NUM;                            //借助下标实现环形  
         sleep(rand()%1);  
     }  
}  
   
void *consumer(void *arg)  
{  
     int i = 0;  
     while (1) {  
         sem_wait(&product_number);                  //消费者将产品数--,为0则阻塞等待  
         printf("-Consume---%d\n", queue[i]);  
         queue[i] = 0;                               //消费一个产品   
         sem_post(&blank_number);                    //消费掉以后,将空格子数++  
   
         i = (i+1) % NUM;  
         sleep(rand()%3);  
     }  
}  
   
int main(int argc, char *argv[])  
{  
     sem_init(&blank_number, 0, NUM);                //初始化空格子信号量为5, 线程间共享--0  
     sem_init(&product_number, 0, 0);                //产品数为0  

     pthread_t pid, cid;  
     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;  
 }

条件变量和信号量实现生产者消费者模型掌握一个即可

Linux系统编程---线程同步

网站公告

今日签到

热门文章

最新发布