[系统运维]linux pthread多线程-2

1.pthread_cancel

??pthread_cancel() 时用来线程退出的，但是在使用的时候有很多注意的地方，甚至是不建议使用。例如：

static void* pthread_func1(void* arg)
{
    while(1)
    {
        printf("haha\n");
        sleep(1);
    }
    return NULL;
}

static void* pthread_func2(void* arg)
{
    int a = 0;
    for( ;; )
        a++;
    return NULL;
}

int main(int argc, char const *argv[]) {
    pthread_t tid1;
    pthread_t tid2;
    pthread_create(&tid1, NULL, pthread_func1, NULL);
    pthread_create(&tid2, NULL, pthread_func2, NULL);

    pthread_cancel(ti1);
    pthread_join(tid1, NULL);
    pthread_cancel(ti2);
    pthread_join(tid2, NULL);
    return 0;
}

??执行时会发现线程一正常的退出，但是线程二还是一直在执行的。因为pthread_cancel是给线程1发出cancel信号，而不是直接退出。但如何处理Cancel信号则由目标线程自己决定，或者忽略、或者立即终止、或者继续运行至Cancelation-point（取消点），由不同的Cancelation状态决定。 线程对于cancel信号的反应可以由pthread_setcancelstate()函数来指定，例如

pthread_setcancelstate(PTHREAD_CANCEL_DISABLE,NULL)

??PTHREAD_CANCEL_DISABLE则是对cancel信号不响应。
??而线程的取消点，则是指应用代码进入内核的系统调用时，例如pthread_join()、pthread_testcancel()、pthread_cond_wait()、pthread_cond_timedwait()、sem_wait()、sigwait()等函数以及read()、write()等会引起阻塞的系统调用都是Cancelation-point，特别是最常见的sleep函数会进入系统调用执行进程切换，也会是取消点。由于例子中线程2全是C语言的代码，所以不会取消线程，特别是如果线程执行pthread_mutex_lock会导致死锁，所以pthread_cancel要慎用。

2.线程同步控制-互斥锁

??第一步要定义互斥量，互斥量结构体为pthread_mutex_t

pthread_mutex_t mutex

??互斥量结构体对象初始化有两种方式

1.pthread_mutex_init(&mutex,NULL);
2.pthread_mutex_t mutex = PTHREAD_MUTEX_INITALIZER; //调用PTHREAD_MUTEX_INITALIZER宏进行初始化

??互斥量加锁和解锁

int pthread_mutex_lock(pthread_mutex_t *mutex);
int pthread_mutex_unlock(pthread_mutex_t *mutex);

??加锁可以理解为当某个线程执行pthread_mutex_lock后，那么他就拥有了该互斥量，其他的线程如果在执行pthread_mutex_lock来获取同一个互斥量的锁的话，会发生阻塞，直到得到锁的进程执行pthread_mutex_unlock，得到所得线程必须要在逻辑代码结束前执行解锁操作，不然别的线程会发生死锁。
??互斥量非阻塞加锁

int pthread_mutex_trylock(pthread_mutex_t *mutex);
无论哪个线程是否得到该互斥量，当前线程执行完pthread_mutex_trylock都会立即返回。

??互斥量的销毁

int pthread_mutex_destory(pthread_mutex_t *mutex);

??使用互斥量的例子：

#define _GNU_SOURCE 
#include <pthread.h>
#include <stdio.h>
#include <unistd.h>
#include <errno.h>
pthread_mutex_t mutex;//互斥量变量 一般申请全局变量
int Num = 0;//公共临界变量

void *fun1(void *arg)
{
	pthread_mutex_lock(&mutex);//加锁 若有线程获得锁，则会阻塞
	while(Num < 3){
		Num++;
		printf("%s:Num = %d\n",__FUNCTION__,Num);
		sleep(1);
	}
	pthread_mutex_unlock(&mutex);//解锁
	pthread_exit(NULL);//线程退出 pthread_join会回收资源
}
void *fun2(void *arg)
{
	pthread_mutex_lock(&mutex);//加锁 若有线程获得锁，则会阻塞
	while(Num > -3){
		Num--;
		printf("%s:Num = %d\n",__FUNCTION__,Num);
		sleep(1);
	}
	pthread_mutex_unlock(&mutex);//解锁
	pthread_exit(NULL);//线程退出 pthread_join会回收资源
}
int main()
{
	int ret;
	pthread_t tid1,tid2;
	ret = pthread_mutex_init(&mutex,NULL);//初始化互斥量
	if(ret != 0){
		perror("pthread_mutex_init");
		return -1;
	}
	ret = pthread_create(&tid1,NULL,fun1,NULL);//创建线程1
	if(ret != 0){
		perror("pthread_create");
		return -1;
		}
	ret = pthread_create(&tid2,NULL,fun2,NULL);//创建线程2
	if(ret != 0){
		perror("pthread_create");
		return -1;
	}
	pthread_join(tid1,NULL);//阻塞回收线程1
	pthread_join(tid2,NULL);//阻塞回收线程2
	pthread_mutex_destroy(&mutex);//销毁互斥量
	return 0;
}

3.线程同步控制-读写锁

// 声明一个读写锁
pthread_rwlock_t rwlock;
...
// 在读之前加读锁
pthread_rwlock_rdlock(&rwlock);

... 共享资源的读操作

// 读完释放锁
pthread_rwlock_unlock(&rwlock);

// 在写之前加写锁
pthread_rwlock_wrlock(&rwlock); 

... 共享资源的写操作

// 写完释放锁
pthread_rwlock_unlock(&rwlock);

// 销毁读写锁
pthread_rwlock_destroy(&rwlock);
//非阻塞请求锁
 pthread_rwlock_tryrdlock(&rwlock)
 pthread_rwlock_trywrlock(&rwlock)

??读写锁是一个锁而不是两个，但是它的特性是读共享但写独占。读写不能同时进行。
?? 例如有A、B、C三个线程：

1. A写锁成功，B请求读锁：B读阻塞
2. A读锁成功，B请求写锁：B写阻塞
3. A读锁成功，B请求读锁：B也读锁成功
4. A读锁成功，B请求写锁同时C请求加锁：B、C都阻塞，A解锁后会先执行B写锁，B解锁后在执行C读锁
5. A读锁成功，B请求读锁后C请求写锁：写锁的优先级比读锁高，A解锁后会先执行C写锁

??读写锁关注的是多个线程中需要允许多个读操作并行的情况，除此之外执行互斥锁比较好

4.线程同步控制-自旋锁

// 声明一个自旋锁变量
pthread_spinlock_t spinlock;
// 初始化   
pthread_spin_init(&spinlock, 0);
// 加锁  
pthread_spin_lock(&spinlock);
// 解锁 
pthread_spin_unlock(&spinlock);
// 销毁  
pthread_spin_destroy(&spinlock);

??自旋锁和互斥锁使用方法类似，只不过互斥锁是阻塞等待，但是自旋锁会一直请求，占用CPU。

5.线程同步控制-信号量

??互斥量用来防止多个线程同时访问临界资源。例如某个硬件设备同一时间只允许一个线程进行操作。而信号量则是起通知作用，例如线程A在等待某件事，线程B完成了则可以给线程A发信号。
??信号量结构体sem_t，信号量初始化：

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

??第三个参数表示信号量的初始值，0代表阻塞，1代表运行。
??信号量的P/V操作：

int sem_wait(sem_t *sem);
int sem_post(sem_t *sem);
``
sem_post函数会释放指定信号量的资源，执行“sem+1”操作，sem_wait函数作用为检测指定信号量是否有资源可用，即sem是否大于0，若大于0则会执行“sem-1”的操作，若为0则会阻塞，直到sem大于0才继续执行。
信号量的销毁：

```c
int sem_destory(sem_t *sem);

#include <stdio.h>
#include <unistd.h>
#include <string.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <dirent.h>
#include <stdlib.h>
#include <pthread.h>
#include <semaphore.h>

sem_t sem; //信号量结构

int data;
/*
线程工作函数
*/
void *thread_work_func(void *dev)
{
    while(1)
    {
        sem_wait(&sem); //获取信号量. 当信号量的值大于0才能获取成功. --
        printf("data=%d\n",data);
        sem_post(&sem); //释放信号量.  ++
        sleep(1);
    }
}

/*
线程工作函数
*/
void *thread_work_func2(void *dev)
{
    while(1)
    {
        sem_wait(&sem); //获取信号量. 当信号量的值大于0才能获取成功. --
        data++;
        sem_post(&sem); //释放信号量.  ++
        sleep(1);
    }
}

int main(int argc,char **argv)
{   
    //初始化信号量
    sem_init(&sem,0,1);

    /*1. 创建子线程1*/
    pthread_t thread_id;
    if(pthread_create(&thread_id,NULL,thread_work_func,NULL)!=0)
    {
        printf("子线程1创建失败.\n");
        return -1;
    }
    /*2. 创建子线程2*/
    pthread_t thread_id2;
    if(pthread_create(&thread_id2,NULL,thread_work_func2,NULL)!=0)
    {
        printf("子线程2创建失败.\n");
        return -1;
    }

    /*3. 等待线程的介绍*/
    pthread_join(thread_id,NULL);
    pthread_join(thread_id2,NULL);

    //销毁信号量
    sem_destroy(&sem);
    return 0;
}

#define _GNU_SOURCE 
#include <pthread.h>
#include <stdio.h>
#include <unistd.h>
#include <errno.h>
#include <semaphore.h>
sem_t sem1,sem2,sem3;//申请的三个信号量变量
void *fun1(void *arg)
{
	sem_wait(&sem1);//因sem1本身有资源，所以不被阻塞 获取后sem1-1 下次会会阻塞
	printf("%s:Pthread Come!\n",__FUNCTION__);
	sem_post(&sem2);// 使得sem2获取到资源
	pthread_exit(NULL);
}

void *fun2(void *arg)
{
	sem_wait(&sem2);//因sem2在初始化时无资源会被阻塞，直至14行代码执行 不被阻塞 sem2-1 下次会阻塞
	printf("%s:Pthread Come!\n",__FUNCTION__);
	sem_post(&sem3);// 使得sem3获取到资源
	pthread_exit(NULL);
}

void *fun3(void *arg)
{
	sem_wait(&sem3);//因sem3在初始化时无资源会被阻塞，直至22行代码执行 不被阻塞 sem3-1 下次会阻塞
	printf("%s:Pthread Come!\n",__FUNCTION__);
	sem_post(&sem1);// 使得sem1获取到资源
	pthread_exit(NULL);
}

int main()
{
	int ret;
	pthread_t tid1,tid2,tid3;
	ret = sem_init(&sem1,0,1);  //初始化信号量1 并且赋予其资源
	if(ret < 0){
		perror("sem_init");
		return -1;
	}
	ret = sem_init(&sem2,0,0); //初始化信号量2 让其阻塞
	if(ret < 0){
		perror("sem_init");
		return -1;
	}
	ret = sem_init(&sem3,0,0); //初始化信号3 让其阻塞
	if(ret < 0){
		perror("sem_init");
	return -1;
	}
	ret = pthread_create(&tid1,NULL,fun1,NULL);//创建线程1
	if(ret != 0){
		perror("pthread_create");
		return -1;
	}
	ret = pthread_create(&tid2,NULL,fun2,NULL);//创建线程2
	if(ret != 0){
		perror("pthread_create");
		return -1;
		}
	ret = pthread_create(&tid3,NULL,fun3,NULL);//创建线程3
	if(ret != 0){
		perror("pthread_create");
		return -1;
	}
	/*回收线程资源*/
	pthread_join(tid1,NULL);
		pthread_join(tid2,NULL);
		pthread_join(tid3,NULL);

	/*销毁信号量*/
	sem_destroy(&sem1);
	sem_destroy(&sem2);
	sem_destroy(&sem3);

	return 0;
}

6.线程同步控制-条件变量

??条件变量是一种同步机制，用来通知其他线程条件满足了。一般是用来通知对方共享数据的状态信息，因此条件变量时结合互斥量来使用的。例如生产者和消费者问题，两个不能通知执行所以要加互斥锁，但是消费者执行之前要确保生产者不是0，不然没法消费，同时生产者要确保有地方进行存储，如果满了，则需要消费者先消费。
??创建和销毁条件变量：

// 初始化条件变量
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
int pthread_cond_init(pthread_cond_t *cond, pthread_condattr_t *cond_attr);//cond_attr通常为NULL

// 销毁条件变量
int pthread_cond_destroy(pthread_cond_t *cond);

??等待条件变量，需要结合互斥锁使用：

pthread_mutex_lock(&g_tMutex);
pthread_cond_wait(&g_tConVar, &g_tMutex);  // 如果条件不满足则，会unlock g_tMutex
                                           // 条件满足后被唤醒，会lock g_tMutex
 /* 操作临界资源 */
pthread_mutex_unlock(&g_tMutex);

??pthread_cond_wait一定要放在pthread_mutex_lock和pthread_mutex_unlock之间进行使用，当线程调用pthread_cond_wait()函数时，该函数会阻塞调用线程，直到发出指定条件的信号。会将调用线程放到等待条件的线程列表上，并对互斥量解锁（这样就不会死锁）。当pthread_cond_wait()返回时（即收到信号后），互斥量再次锁住。因为要根据共享变量的状态来决定是否等待（即互斥的访问共享变量），因此在该函数调用前要先获取相关互斥量（即正在互斥的访问该共享变量）。所以该函数必须要放在pthread_mutex_lock和pthread_mutex_unlock之间的临界区内。对于pthread_cond_wait，最好前面使用while而不是if做判断。
??通知条件变量：

int pthread_cond_signal(pthread_cond_t *cond);

??pthread_cond_signal函数只会唤醒一个等待cond条件变量的线程，示例代码如下：

pthread_cond_signal(&g_tConVar);

例程：

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

#define MAX 5//共享变量最大值上界
int num;//全局共享变量

pthread_mutex_t * m;//互斥量
pthread_cond_t * push;//条件变量push:表示生产者生产了一个产品
pthread_cond_t * pop;//条件变量pop:表示消费者消费了一个产品

//消费者线程入口函数
void* consumer(void*arg){
    for(int i = 0 ; i < 15 ; ++i){
        pthread_mutex_lock(m);//上锁
        if(num == 0){
            pthread_cond_wait(push,m);//没有产品了，阻塞等待push信号唤醒
        }
        num--;
        printf("the num is %d now\n",num);
        pthread_cond_signal(pop);//消费了一个产品，发出pop信号
        pthread_mutex_unlock(m);//解锁
    }
}
//生产者线程入口函数
void* producer(void*arg){
    for(int i = 0 ; i < 15 ; ++i){
        pthread_mutex_lock(m);//上锁
        if(num == MAX){
            pthread_cond_wait(pop,m);//产品满了，阻塞等待pop信号唤醒
        }
        num++;
        printf("the num is %d now\n",num);
        pthread_cond_signal(push);//生产了一个产品，发出push信号
        pthread_mutex_unlock(m);//解锁
    }
}

int main(){
    m = (pthread_mutex_t*)malloc(sizeof(pthread_mutex_t));
    
    push = (pthread_cond_t*)malloc(sizeof(pthread_cond_t));
    pop = (pthread_cond_t*)malloc(sizeof(pthread_cond_t));
    
    pthread_mutex_init(m,NULL);//初始化互斥量
    pthread_cond_init(push,NULL);//初始化条件变量
    pthread_cond_init(pop,NULL);
    
    pthread_t producerThread[20];
    for(int i = 0 ; i < 20 ; ++i){//创建20个生产者线程
        pthread_create(&producerThread[i],NULL,producer,NULL);
    }
    
    pthread_t consumerThread[20];
    for(int i = 0 ; i < 20 ; ++i){//创建20个消费者线程
        pthread_create(&consumerThread[i],NULL,consumer,NULL);
    }
    
    void* status;
    for(int i = 0 ; i < 20 ; ++i){
        pthread_join(producerThread[i],&status);//阻塞等待线程终止
        pthread_join(consumerThread[i],&status);
    }

    printf("\nthe num finally is %d\n",num);
}

??执行这个程序会导致有可能出现num 大于max的情况，例如多个消费者阻塞，当生产者发出一个信号后，多个消费者被唤醒，由于是在if里面，则他们都会接着往下执行并且num是全局变量，所以会出现num大于max的情况，如果改为wile则不会发生，wait被唤醒后会再次执行一次判断。

//消费者线程入口函数
void* consumer(void*arg){
    for(int i = 0 ; i < 15 ; ++i){
        pthread_mutex_lock(m);//上锁
        while(num == 0){//循环判断资源是否可用
            pthread_cond_wait(push,m);//没有产品了，阻塞等待push信号唤醒
        }
        num--;
        printf("the num is %d now\n",num);
        pthread_cond_signal(pop);//消费了一个产品，发出pop信号
        pthread_mutex_unlock(m);//解锁
    }
}
//生产者线程入口函数
void* producer(void*arg){
    for(int i = 0 ; i < 15 ; ++i){
        pthread_mutex_lock(m);//上锁
        while(num == MAX){//循环判断资源是否可用
            pthread_cond_wait(pop,m);//产品满了，阻塞等待pop信号唤醒
        }
        num++;
        printf("the num is %d now\n",num);
        pthread_cond_signal(push);//生产了一个产品，发出push信号
        pthread_mutex_unlock(m);//解锁
    }
}

2022深度学习开发者峰会

5月20日13:00让我们相聚云端，共襄盛会！