八、Linux系统编程：多线程同步

8 多线程同步

多线程同步主要有信号量、互斥量、条件变量和读写锁四种方式。

8.1 背景

实例：数据竞争

#include 
#include 
#include 
//线程并发执行
using namespace std;
void* handle(void* p){
    int* pn = (int*)p;
    for(int i = 0;i < 10;++i){
        sleep(1);
        cout << pthread_self() << ":" << --*pn << endl;
    }
    return NULL;
}
int main(){
    //每个进程里默认有一个线程  
    //主线程
    cout << getpid() << endl;
    cout << pthread_self() << endl;

    int n = 0;
    pthread_t tid;
    pthread_create(&tid,NULL,handle,&n); //创建一个子线程
    for(int i = 0;i < 5;++i){
        sleep(1);
        cout << pthread_self() << ":" << ++n << endl;
    }
    pthread_join(tid,NULL);
}

8.2 信号量

操作	函数
创建	int sem_init(sem_t *sem, int pshared, unsigned int value)
销毁	int sem_destroy(sem_t *sem)
阻塞等待	int sem_wait(sem_t *sem)
非阻塞等待	int sem_trywait(sem_t * sem)
触发	int sem_post(sem_t *sem)

8.2.1 创建

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

参数	含义
sem	信号量对象
pshared	信号量类型。0：线程共享；<0：进程共享
value	初始值

返回值

返回值	含义
0	成功
-1	失败

8.2.2 销毁

int sem_destroy(sem_t *sem)

sem：信号量对象
返回值

返回值	含义
0	成功
-1	失败

8.2.3 阻塞等待

int sem_wait(sem_t *sem)

sem：信号量对象
返回值

返回值	含义
0	成功
-1	失败

8.2.4 非阻塞等待

int sem_trywait(sem_t * sem)

sem：信号量对象
返回值

返回值	含义
0	成功
-1	失败

8.2.5 触发

int sem_post(sem_t *sem)

sem：信号量对象
返回值

返回值	含义
0	成功
-1	失败

8.2.6 实例：加入信号量解决数据竞争

#include 
#include 
#include 
#include 
//线程并发
sem_t sem;
using namespace std;
void* handle(void* p) {
    int* pn = (int*)p;
    for(int i = 0; i < 10; i++) {
        sleep(1);
	sem_wait(&sem);
        cout << pthread_self() << ":" << --*pn << endl;
        sem_post(&sem);
    }
    delete pn;
    return NULL;
}
pair test() {
    // int* p = new int(0); //p存放默认0
    int* p = new int; //p存放默认0
    pthread_t tid;
    pthread_create(&tid,NULL,handle,p);
    //return pair(tid,p);
    return {tid,p};
}
int main() { // 主线程
    sem_init(&sem,0,1);
    cout << getpid() << endl;
    cout << pthread_self() << endl;
    
    auto [tid,p] = test();//C++17
    for(int i = 0; i < 5; i++) {
        sleep(1);
	sem_wait(&sem);
        cout << pthread_self() << ":" << ++*p << endl;
	sem_post(&sem);
    }
    pthread_join(tid,NULL);
    sem_destroy(&sem);
    return 0;
}

[root@localhost 6]# ./a.out 
6200
140180595820352
140180595820352:1
140180577863424:0
140180595820352:1
140180577863424:0
140180595820352:1
140180577863424:0
140180595820352:1
140180577863424:0
140180595820352:1
140180577863424:0
140180577863424:-1
140180577863424:-2
140180577863424:-3
140180577863424:-4
140180577863424:-5

8.3 互斥量 8.3.1 分类

分类	实现	特点
静态分配互斥量	pthread_mutex_t mutex= PTHREAD_MUTEX_INITIALIZER;	简单
动态分配互斥量	pthread_mutex_init(&mutex, NULL);pthread_mutex_destroy(&mutex);	可以设置更多的选项

8.3.2 操作

操作	函数
加锁	int pthread_mutex_lock(pthread_t *mutex)
尝试加锁	int pthread_mutex_trylock(pthread_t *mutex)
解锁	int pthread_mutex_unlock(pthread_t *mutex)

参数
mutex ：互斥锁

8.3.3 实例：加入互斥锁解决数据竞争

#include 
#include 
#include 
#include 
//线程并发
using namespace std;
pthread_mutex_t mutex;
void* handle(void* p) {
    int* pn = (int*)p;
    for(int i = 0; i < 10; i++) {
        sleep(1);
	//sem_wait(&sem);
        pthread_mutex_lock(&mutex);
		cout << pthread_self() << ":" << --*pn << endl;
	//sem_post(&sem);
        pthread_mutex_unlock(&mutex);
    }
    delete pn;
    return NULL;
}
pair test() {
    // int* p = new int(0); //p存放默认0
    int* p = new int; //p存放默认0
    pthread_t tid;
    pthread_create(&tid,NULL,handle,p);
    //return pair(tid,p);
    return {tid,p};
}
int main() { // 主线程
    //sem_init(&sem,0,1);
    pthread_mutex_init(&mutex,NULL);
    cout << getpid() << endl;
    cout << pthread_self() << endl;
    
    auto [tid,p] = test();//C++17
    for(int i = 0; i < 5; i++) {
        sleep(1);
	//sem_wait(&sem);
        pthread_mutex_lock(&mutex);
        cout << pthread_self() << ":" << ++*p << endl;
	//sem_post(&sem);
        pthread_mutex_unlock(&mutex);
    }
    pthread_join(tid,NULL);
    //sem_destroy(&sem);
    pthread_mutex_destroy(&mutex);
    return 0;
}

5949
140699829409600
140699829409600:1
140699811452672:0
140699829409600:1
140699811452672:0
140699829409600:1
140699811452672:0
140699829409600:1
140699811452672:0
140699829409600:1
140699811452672:0
140699811452672:-1
140699811452672:-2
140699811452672:-3
140699811452672:-4
140699811452672:-5

8.4 信号量与互斥量的区别

区别	信号量	互斥量
使用对象	线程和进程	线程
量值	非负整数	0或1
操作	PV操作可由不同线程完成	加锁和解锁必须由同一线程使用
应用	用于线程的同步	用于线程的互斥

互斥：主要关注于资源访问的唯一性和排他性。
同步：主要关注于操作的顺序，同步以互斥为前提。

8.5 条件变量

概念
线程挂起直到共享数据的某些条件得到满足

8.5.1 分类

分类	实现	特点
静态分配条件变量	pthread_cond_t cond = PTHREAD_COND_INITIALIZER;	简单
动态分配静态变量	pthread_cond_init(&cond, NULL); pthread_cond_destroy(&cond);	可以设置更多的选项

8.5.2 操作

操作	函数
条件等待	int pthread_cond_wait(pthread_cond_t cond, pthread_mutex_t mutex)
计时等待	int pthread_cond_timedwait(pthread_cond_t cond, pthread_mutex_t mutex, const struct timespec *abstime)
单个激活	int pthread_cond_signal(pthread_cond_t *cond)
全部激活	int pthread_cond_broadcast(pthread_cond_t *cond)

8.5.2.1 条件等待

int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex)

参数

参数	含义
cond	条件变量
mutex	互斥锁

8.5.2.2 计时等待

int pthread_cond_timedwait(pthread_cond_t *cond, pthread_mutex_t *mutex, const struct timespec *abstime)

参数

参数	含义
cond	条件变量
mutex	互斥锁
abstime	等待时间

返回值
ETIMEDOUT ：超时结束等待

8.5.2.3 单个激活

int pthread_cond_signal(pthread_cond_t *cond)

参数
cond ：条件变量
返回值

返回值	含义
0	成功
正数	错误码

8.5.2.4 全部激活

int pthread_cond_broadcast(pthread_cond_t *cond)

参数
cond ：条件变量
返回值

返回值	含义
0	成功
正数	错误码

8.5.3 条件变量一般与互斥锁一起使用

为什么一起使用？

条件变量的作用
1、条件变量的作用是用于多线程之间的线程同步。
2、线程同步是指线程间需要按照预定的先后顺序进行的行为，比如我想要线程1完成了某个步骤之后，才允许线程2开始工作，这个时候就可以使用条件变量来达到目的。
互斥锁的作用
1、互斥锁主要是用在多线程编程时，多个线程同时访问同一个变量的情况下，保证在某一个时刻只能有一个线程访问。
2、每个线程在访问共享变量的时候，首先要先获得锁，然后才能访问共享变量，当一个线程成功获得锁时，其他变量都会block在获取锁这一步，这样就达到了保护共享变量的目的。
配合使用
1、保证在任何时候都只有一个线程在等待条件满足，并且当条件满足后等待的线程会收到通知，而不必轮寻，同时释放锁。
2、条件变量就是减少竞争，在条件满足后就去唤醒等待的线程。
例子
条件变量配合着互斥锁使用，是为了解决：在线程A中调用了 pthread_cond_wait，但是线程A还没有进入wait cond的时候，线程B调用了pthread_cond_signal，如果不用mutex，这个cond_signal就丢失了，而如果加上了锁，则线程B必须要等到mutex被释放了的时候才能调用pthread_cond_signal。
总结
条件变量用在某个线程需要在某种条件才去保护它将要操作的临界区的情况下，从而避免了线程不断轮询检查该条件是否成立而降低效率的情况，这是实现了效率提高。

8.5.4 实例：条件变量与互斥锁一起使用

#include 
#include 
#include 
#include 
#include 
#include 
using namespace std;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;

void* sum(void* data){
     vector& p = *(vector*)data;
     while(true){
     	pthread_mutex_lock(&mutex);
		cout << "sum lock" << endl;
		if(p.size() < 5){
	    	cout << "sum before wait" << endl;
	    	pthread_cond_wait(&cond,&mutex); //即具有解锁功能，又具有加锁功能
	    	cout << "sum afer wait" << endl;
		}
		cout << "sum:" << accumulate(p.begin(),p.end(),0) << endl;
		p.clear();
		pthread_mutex_unlock(&mutex);
		cout << "sum unlock" << endl;
     }
}
int main(){
    vector data;
    pthread_t tid;
    pthread_create(&tid,NULL,sum,&data);    
    int n;
    while(cin >> n){
     	pthread_mutex_lock(&mutex);
		cout << "main lock" << endl;
		data.push_back(n);
        if(data.size() >= 5){
	   		pthread_cond_signal(&cond);
        	cout << "main signal" << endl;	    
		}	
		pthread_mutex_unlock(&mutex);
		cout << "main unlock" << endl;
    }
    pthread_join(tid,NULL);
}

[root@localhost 7]# ./a.out 
sum lock
sum before wait
1
main lock
main unlock
2
main lock
main unlock
3
main lock
main unlock
4
main lock
main unlock
5
main lock
main signal
main unlock
sum afer wait
sum:15
sum unlock
sum lock
sum before wait

8.6 读写锁

资源访问分为两种情况：读操作和写操作。

读写锁比互斥锁（mutex）有更高的适用性，可以多个线程同时占用读模式的读写锁，但是只能一个线程占用写模式的读写锁。

1、当读写锁是写加锁状态时，在这个锁被解锁之前，所有试图对这个锁加锁的线程都会被阻塞；
2、当读写锁在读加锁状态时，所有试图以读模式对它进行加锁的线程都可以得到访问权，但是以写模式对它进行枷锁的线程将阻塞；
3、当读写锁在读模式锁状态时，如果有另外线程试图以写模式加锁，读写锁通常会阻塞随后的读模式锁请求，这样可以避免读模式锁长期占用，而等待的写模式锁请求长期阻塞；
这种锁适用对数据结构进行读的次数比写的次数多的情况下，因为可以进行读锁共享。

比喻
新闻发布会
领导发言与聊天
概念
共享独占
读取锁(共享)
写入锁(独占)

8.6.1 分类

分类	实现	特点
静态分配读写锁	pthread_rwlock_t rwlock = PTHREAD_RWLOCK_INITIALIZER	简单
动态分配读写锁	pthread_rwlock_init(&rwlock, NULL); pthread_rwlock_destroy(&rwlock);	可以设置更多的选项

8.6.2 操作

读锁

int pthread_rwlock_rdlock(pthread_rwlock_t *rwlock);
int pthread_rwlock_tryrdlock(pthread_rwlock_t *rwlock);

写锁

int pthread_rwlock_wrlock(pthread_rwlock_t *rwlock);
int pthread_rwlock_trywrlock(pthread_rwlock_t *rwlock);

解锁

int pthread_rwlock_unlock(pthread_rwlock_t *rwlock)

8.6.3 实例：火车票的查询与购买

#include 
#include 
#include 
#include 
pthread_rwlock_t rwlock = PTHREAD_RWLOCK_INITIALIZER;
int count = 10000;
int put_cur = 0;
void* search(void* arg){
    for(;;){
        pthread_rwlock_rdlock(&rwlock);
        printf("leave %dn",count); 
        usleep(500000);
        pthread_rwlock_unlock(&rwlock);
    }
}
void rollback(void* arg){
    count -= put_cur;
    printf("rollback %d  to %dn",put_cur,count);
    pthread_rwlock_unlock(&rwlock);
}
void* put(void* arg){
    pthread_cleanup_push(rollback,NULL);
    for(;;){
        pthread_rwlock_wrlock(&rwlock);
        put_cur = rand()%1000;
        count += put_cur;
        printf("put %d ok,leave %dn",put_cur,count);
        usleep(500000);
        pthread_rwlock_unlock(&rwlock);
    }
    pthread_cleanup_pop(0);
}
void* hacker(void* arg){
    sleep(2);
    pthread_t* ptids = (pthread_t*)arg;
    pthread_cancel(ptids[0]);
    printf("33[41;34mcancel %lu33[0mn",ptids[0]);
    return NULL;
}
void* get(void* arg){
    for(;;){
        pthread_rwlock_wrlock(&rwlock);
        int cur = rand()%1000;
        if(count>cur){
            count -= cur;
            printf("crash %d leave %dn",cur,count);
        }else{
            printf("leave not enought %dn",count);
        }
        usleep(500000);
        pthread_rwlock_unlock(&rwlock);
    }
}
 
int main(){
    pthread_t tids[4];
    typedef void*(*func_t)(void*);
    func_t funcs[4] = {put,get,search,hacker};
    int i=0;
    pthread_setconcurrency(4);
    for(i=0;i<4;i++){
        pthread_create(&tids[i],NULL,funcs[i],tids);
    }
    for(i=0;i<4;i++){
        pthread_join(tids[i],NULL);
    }
}

四个线程：一个查看、一个放票、一个取票、一个随机破坏。

8.7 C++创建多线程

#include 
#include 
#include 
using namespace std;

int main() {
    mutex m;
    thread t([&m]() {
        for(int i = 0; i < 10; ++i) {
            this_thread::sleep_for(500ms);
            m.lock();
            cout << this_thread::get_id() << ":" << i << endl;
            m.unlock();
        }
    });
    for(int i = 0; i < 10; ++i) {
        this_thread::sleep_for(500ms);
        m.lock();
        cout << this_thread::get_id() << ":" << i << endl;
        m.unlock();
    }
    t.join();
}

[root@localhost 7]# g++ threadcxx.cpp -pthread
[root@localhost 7]# ./a.out 
140438391899968:0
140438373943040:0
140438391899968:1
140438373943040:1
140438391899968:2
140438373943040:2
140438391899968:3
140438373943040:3
140438391899968:4
140438373943040:4
140438391899968:5
140438373943040:5
140438391899968:6
140438373943040:6
140438391899968:7
140438373943040:7
140438391899968:8
140438373943040:8
140438391899968:9
140438373943040:9

八、Linux系统编程：多线程同步

C/C++/C#相关栏目本月热门文章