线程同步
1. 线程同步概念
线程同步是指多个线程协调它们的执行顺序,以确保它们正确、安全地访问共享资源。在并发编程中,当多个线程同时访问共享数据或资源时,可能会导致竞争条件(Race Condition)和其他并发问题
所谓的同步并不是多个线程同时对内存进行访问,而是按照先后顺序依次进行的。
1.1 为什么要同步
在研究线程同步之前,先来看一个两个线程交替数数(每个线程数50个数,交替数到100)的例子:
#include <stdio.h>
#include <unistd.h>
#include <pthread.h>
#include <mutex>
#define MAX 20
// 全局变量
int number;
// 线程处理函数
void* funcA_num(void* arg)
{
for(int i=0; i<MAX; ++i)
{
int cur = number;
cur++;
sleep(1);
number = cur;
printf("Thread A, id = %lu, number = %d\n", pthread_self(), number);
}
return NULL;
}
void* funcB_num(void* arg)
{
for(int i=0; i<MAX; ++i)
{
int cur = number;
cur++;
number = cur;
printf("Thread B, id = %lu, number = %d\n", pthread_self(), number);
sleep(1);
}
return NULL;
}
int main(int argc, const char* argv[])
{
pthread_t p1, p2;
// 创建两个子线程
pthread_create(&p1, NULL, funcA_num, NULL);
pthread_create(&p2, NULL, funcB_num, NULL);
// 阻塞,资源回收
pthread_join(p1, NULL);
pthread_join(p2, NULL);
return 0;
}
运行程序的结果为:
Thread B, id = 2, number = 1
Thread A, id = 1, number = 1
Thread B, id = 2, number = 2
Thread A, id = 1, number = 3
Thread B, id = 2, number = 4
Thread B, id = 2, number = 5
Thread A, id = 1, number = 5
Thread A, id = 1, number = 6
Thread B, id = 2, number = 7
Thread A, id = 1, number = 8
Thread B, id = 2, number = 9
Thread A, id = 1, number = 10
Thread B, id = 2, number = 11
Thread B, id = 2, number = 12
Thread A, id = 1, number = 12
Thread B, id = 2, number = 13
Thread A, id = 1, number = 13
Thread A, id = 1, number = 14
Thread B, id = 2, number = 15
Thread B, id = 2, number = 16
Thread A, id = 1, number = 16
Thread A, id = 1, number = 17
Thread B, id = 2, number = 18
Thread B, id = 2, number = 19
Thread A, id = 1, number = 19
Thread B, id = 2, number = 20
Thread A, id = 1, number = 20
Thread A, id = 1, number = 21
Thread B, id = 2, number = 22
Thread A, id = 1, number = 23
Thread B, id = 2, number = 23
Thread B, id = 2, number = 24
Thread A, id = 1, number = 24
Thread A, id = 1, number = 25
Thread B, id = 2, number = 26
Thread A, id = 1, number = 27
Thread B, id = 2, number = 28
Thread A, id = 1, number = 29
Thread B, id = 2, number = 30
Thread A, id = 1, number = 31
可以看出每个线程各数20个数, 最后到31就结束了(每次运行结果也不一样), 其原因就是没有对线程进行同步处理,造成了数据的混乱。就是说一个线程对数据做++操作后, 还没等把数据写入到内存之后, 另一个线程就开始工作了, 拿到的数字还是做加法操作之前的数字, 就会造成数据混乱
1.2 同步方式
常用的线程同步方式有四种:**互斥锁、读写锁、条件变量、信号量。**所谓的共享资源就是多个线程共同访问的变量,这些变量通常为全局数据区变量或者堆区变量,这些变量对应的共享资源也被称之为临界资源。
2. 互斥锁
互斥锁是线程同步最常用的一种方式,通过互斥锁可以锁定一个代码块, 被锁定的这个代码块, 所有的线程只能顺序执行(不能并行处理),这样多线程访问共享资源数据混乱的问题就可以被解决了,需要付出的代价就是执行效率的降低,因为默认临界区多个线程是可以并行处理的,现在只能串行处理。
2.1 相关函数
在Linux中互斥锁的类型为pthread_mutex_t
,创建一个这种类型的变量就得到了一把互斥锁:
pthread_mutex_t mutex;
一般情况下,每一个共享资源对应一个把互斥锁,锁的个数和线程的个数无关。
锁的初始化和销毁:
// 初始化互斥锁
// restrict: 是一个关键字, 用来修饰指针, 只有这个关键字修饰的指针可以访问指向的内存地址, 其他指针是不行的
int pthread_mutex_init(pthread_mutex_t *restrict mutex,
const pthread_mutexattr_t *restrict attr);
// 释放互斥锁资源
int pthread_mutex_destroy(pthread_mutex_t *mutex);
参数:
- mutex: 互斥锁变量的地址
- attr: 互斥锁的属性, 一般使用默认属性即可, 这个参数指定为NULL
互斥锁的锁定和解锁
// 修改互斥锁的状态, 将其设定为锁定状态, 这个状态被写入到参数 mutex 中
int pthread_mutex_lock(pthread_mutex_t *mutex);
// 对互斥锁解锁
int pthread_mutex_unlock(pthread_mutex_t *mutex);
对于加锁函数pthread_mutex_lock
:
- 没有被锁定, 是打开的, 这个线程可以加锁成功, 这个这个锁中会记录是哪个线程加锁成功了
- 如果被锁定了, 其他线程加锁就失败了, 这些线程都会阻塞在这把锁上
- 当这把锁被解开之后, 这些阻塞在锁上的线程就解除阻塞了,并且这些线程是通过竞争的方式对这把锁加锁,没抢到锁的线程继续阻塞
解锁函数: 不是所有的线程都可以对互斥锁解锁,哪个线程加的锁, 哪个线程才能解锁成功。
尝试加锁函数
// 尝试加锁
int pthread_mutex_trylock(pthread_mutex_t *mutex);
调用这个函数对互斥锁变量加锁还是有两种情况:
- 如果这把锁没有被锁定是打开的,线程加锁成功
- 如果锁变量被锁住了,调用这个函数加锁的线程,不会被阻塞,加锁失败直接返回错误号
2.2 互斥锁的使用
在看上面的例子, 加上互斥锁之后:
//
// Created by kk on 2024/1/11.
//
#include <stdio.h>
#include <unistd.h>
#include <pthread.h>
#include <mutex>
#define MAX 20
// 全局变量
int number;
// 定义一把锁
pthread_mutex_t mutex;
// 线程处理函数
void* funcA_num(void* arg)
{
for(int i=0; i<MAX; ++i)
{
pthread_mutex_lock(&mutex);
int cur = number;
cur++;
sleep(1);
number = cur;
pthread_mutex_unlock(&mutex);
printf("Thread A, id = %lu, number = %d\n", pthread_self(), number);
}
return NULL;
}
void* funcB_num(void* arg)
{
for(int i=0; i<MAX; ++i)
{
pthread_mutex_lock(&mutex);
int cur = number;
cur++;
number = cur;
pthread_mutex_unlock(&mutex);
printf("Thread B, id = %lu, number = %d\n", pthread_self(), number);
sleep(1);
}
return NULL;
}
int main(int argc, const char* argv[])
{
pthread_t p1, p2;
// 初始化互斥锁
pthread_mutex_init(&mutex, NULL);
// 创建两个子线程
pthread_create(&p1, NULL, funcA_num, NULL);
pthread_create(&p2, NULL, funcB_num, NULL);
// 阻塞,资源回收
pthread_join(p1, NULL);
pthread_join(p2, NULL);
// 销毁互斥锁
pthread_mutex_destroy(&mutex);
return 0;
}
运行结果为:
Thread A, id = 1, number = 1
Thread B, id = 2, number = 2
Thread A, id = 1, number = 3
Thread B, id = 2, number = 4
Thread A, id = 1, number = 5
Thread B, id = 2, number = 6
Thread A, id = 1, number = 7
Thread B, id = 2, number = 8
Thread A, id = 1, number = 9
Thread B, id = 2, number = 10
Thread A, id = 1, number = 12
Thread B, id = 2, number = 12
Thread B, id = 2, number = 14
Thread A, id = 1, number = 13
Thread A, id = 1, number = 15
Thread B, id = 2, number = 16
Thread A, id = 1, number = 18
Thread B, id = 2, number = 18
Thread A, id = 1, number = 19
Thread B, id = 2, number = 20
Thread B, id = 2, number = 22
Thread A, id = 1, number = 22
Thread A, id = 1, number = 23
Thread B, id = 2, number = 24
Thread A, id = 1, number = 25
Thread B, id = 2, number = 26
Thread A, id = 1, number = 27
Thread B, id = 2, number = 28
Thread B, id = 2, number = 30
Thread A, id = 1, number = 30
Thread A, id = 1, number = 31
Thread B, id = 2, number = 32
Thread A, id = 1, number = 33
Thread B, id = 2, number = 34
Thread A, id = 1, number = 35
Thread B, id = 2, number = 36
Thread A, id = 1, number = 37
Thread B, id = 2, number = 38
Thread A, id = 1, number = 39
Thread B, id = 2, number = 40
3. 死锁
死锁是指两个或多个线程或进程在执行过程中,因争夺资源而造成的一种互相等待的状态,导致程序无法继续执行下去。如果线程死锁造成的后果是:所有的线程都被阻塞,并且线程的阻塞是无法解开的(因为可以解锁的线程也被阻塞了)。
3.1 造成死锁的几种场景
- 加锁后忘记解锁
// 场景1
void func()
{
for(int i=0; i<6; ++i)
{
// 当前线程A加锁成功, 当前循环完毕没有解锁, 在下一轮循环的时候自己被阻塞了
// 其余的线程也被阻塞
pthread_mutex_lock(&mutex);
....
.....
// 忘记解锁
}
}
// 场景2
void func()
{
for(int i=0; i<6; ++i)
{
// 当前线程A加锁成功
// 其余的线程被阻塞
pthread_mutex_lock(&mutex);
....
.....
if(xxx)
{
// 函数退出, 没有解锁(解锁函数无法被执行了)
return ;
}
pthread_mutex_lock(&mutex);
}
}
28行, 会直接return 没有解锁
- 重复加锁, 造成死锁
void func()
{
for(int i=0; i<6; ++i)
{
// 当前线程A加锁成功
// 其余的线程阻塞
pthread_mutex_lock(&mutex);
// 锁被锁住了, A线程阻塞
pthread_mutex_lock(&mutex);
....
.....
pthread_mutex_unlock(&mutex);
}
}
// 隐藏的比较深的情况
void funcA()
{
for(int i=0; i<6; ++i)
{
// 当前线程A加锁成功
// 其余的线程阻塞
pthread_mutex_lock(&mutex);
....
.....
pthread_mutex_unlock(&mutex);
}
}
void funcB()
{
for(int i=0; i<6; ++i)
{
// 当前线程A加锁成功
// 其余的线程阻塞
pthread_mutex_lock(&mutex);
funcA(); // 重复加锁
....
.....
pthread_mutex_unlock(&mutex);
}
}
37行 再次调用funcA()就会重复加锁
- 在程序中有多个共享资源, 因此有很多把锁,随意加锁,导致相互被阻塞
场景描述:
1. 有两个共享资源:X, Y,X对应锁A, Y对应锁B
- 线程A访问资源X, 加锁A
- 线程B访问资源Y, 加锁B
2. 线程A要访问资源Y, 线程B要访问资源X,因为资源X和Y已经被对应的锁锁住了,因此这个两个线程被阻塞
- 线程A被锁B阻塞了, 无法打开A锁
- 线程B被锁A阻塞了, 无法打开B锁
3.2 产生死锁的四个条件
- 互斥条件;
- 持有并等待条件;
- 不可剥夺条件;
- 环路等待条件;
互斥条件
互斥条件是指多个线程不能同时使用同一个资源。
比如下图,如果线程 A 已经持有的资源,不能再同时被线程 B 持有,如果线程 B 请求获取线程 A 已经占用的资源,那线程 B 只能等待,直到线程 A 释放了资源。
持有并等待条件
持有并等待条件是指,当线程 A 已经持有了资源 1,又想申请资源 2,而资源 2 已经被线程 C 持有了,所以线程 A 就会处于等待状态,但是线程 A 在等待资源 2 的同时并不会释放自己已经持有的资源 1。
不可剥夺条件
不可剥夺条件是指,当线程已经持有了资源 ,在自己使用完之前不能被其他线程获取,线程 B 如果也想使用此资源,则只能在线程 A 使用完并释放后才能获取。
环路等待条件
环路等待条件指的是,在死锁发生的时候,两个线程获取资源的顺序构成了环形链。
比如,线程 A 已经持有资源 2,而想请求资源 1, 线程 B 已经获取了资源 1,而想请求资源 2,这就形成资源请求等待的环形图。
3.3 利用工具排查死锁问题
在 Linux 下,我们可以使用 pstack
+ gdb
工具来定位死锁问题。
pstack 命令可以显示每个线程的栈跟踪信息(函数调用过程),它的使用方式也很简单,只需要 pstack <pid>
就可以了。
那么,在定位死锁问题时,我们可以多次执行 pstack 命令查看线程的函数调用过程,多次对比结果,确认哪几个线程一直没有变化,且是因为在等待锁,那么大概率是由于死锁问题导致的。
4. 读写锁
4.1 概念
读写锁(Read-Write Lock)是一种用于支持多线程并发访问的同步机制,它分为读锁和写锁两种类型。在某些场景中,多个线程可能同时读取共享资源,但只有一个线程能够写入共享资源。读写锁通过允许多个线程同时获取读锁,但只允许一个线程获取写锁,来提高并发性能。
读写锁的工作原理是:
- 当「写锁」没有被线程持有时,多个线程能够并发地持有读锁,这大大提高了共享资源的访问效率,因为「读锁」是用于读取共享资源的场景,所以多个线程同时持有读锁也不会破坏共享资源的数据。
- 但是,一旦「写锁」被线程持有后,读线程的获取读锁的操作会被阻塞,而且其他写线程的获取写锁的操作也会被阻塞。
4.2 相关函数
锁的类型为pthread_rwlock_t,有了类型之后就可以创建一把互斥锁了:
pthread_rwlock_t rwlock;
初始化和销毁
#include <pthread.h>
pthread_rwlock_t rwlock;
// 初始化读写锁
int pthread_rwlock_init(pthread_rwlock_t *restrict rwlock,
const pthread_rwlockattr_t *restrict attr);
// 释放读写锁占用的系统资源
int pthread_rwlock_destroy(pthread_rwlock_t *rwlock);
参数:
- rwlock: 读写锁的地址,传出参数
- attr: 读写锁属性,一般使用默认属性,指定为NULL
加读锁, 锁定读操作
// 在程序中对读写锁加读锁, 锁定的是读操作
int pthread_rwlock_rdlock(pthread_rwlock_t *rwlock);
调用这个函数,如果读写锁是打开的,那么加锁成功;如果读写锁已经锁定了读操作,调用这个函数依然可以加锁成功,因为读锁是共享的;如果读写锁已经锁定了写操作,调用这个函数的线程会被阻塞。
// 在程序中对读写锁加写锁, 锁定的是写操作
int pthread_rwlock_wrlock(pthread_rwlock_t *rwlock);
调用这个函数,如果读写锁是打开的,那么加锁成功;如果读写锁已经锁定了读操作或者锁定了写操作,调用这个函数的线程会被阻塞。
// 解锁, 不管锁定了读还是写都可用解锁
int pthread_rwlock_unlock(pthread_rwlock_t *rwlock);
5. 自旋锁
当已经有一个线程加锁后,其他线程加锁则就会失败,互斥锁和自旋锁对于加锁失败后的处理方式是不一样的:
- 互斥锁加锁失败后,线程会释放 CPU ,给其他线程;
- 自旋锁加锁失败后,线程会忙等待,直到它拿到锁;(一直占用CPU)
互斥锁相比自旋锁会有两次上下文的切换, 这就是多出来的开销成本
- 当线程加锁失败时,内核会把线程的状态从「运行」状态设置为「睡眠」状态,然后把 CPU 切换给其他线程运行;
- 接着,当锁被释放时,之前「睡眠」状态的线程会变为「就绪」状态,然后内核会在合适的时间,把 CPU 切换给该线程运行。
当被锁住的代码执行时间很短的话, 上下文切换的时间可能比代码执行时间还要久
所以,如果你能确定被锁住的代码执行时间很短,就不应该用互斥锁,而应该选用自旋锁,否则使用互斥锁。
一般加锁的过程,包含两个步骤:
- 第一步,查看锁的状态,如果锁是空闲的,则执行第二步;
- 第二步,将锁设置为当前线程持有;
自旋锁会将检查所的状态和加锁两个步骤合并成一条硬件级指令, 原子指令, 这样就保证了这两个步骤是不可分割的,要么一次性执行完两个步骤,要么两个步骤都不执行。
自旋锁与互斥锁使用层面比较相似,但实现层面上完全不同:当加锁失败时,互斥锁用「线程切换」来应对,自旋锁则用「忙等待」来应对。
6. 条件变量
6.1 概述
条件变量(Condition Variable)是一种多线程同步的机制,用于在多个线程之间建立通信。条件变量通常与互斥锁(Mutex)一起使用,以解决线程间的协调和同步问题。
**条件变量用于在某个条件发生或者满足时通知其他线程。**典型的情况是,一个线程等待某个条件变为真,而另一个线程在某些情况下负责将条件设置为真,并通知等待的线程。这样,等待线程可以被唤醒并继续执行。
6.2 相关函数
条件变量类型对应的类型为pthread_cond_t,这样就可以定义一个条件变量类型的变量了:
pthread_cond_t cond;
初始化和销毁
#include <pthread.h>
pthread_cond_t cond;
// 初始化
int pthread_cond_init(pthread_cond_t *restrict cond,
const pthread_condattr_t *restrict attr);
// 销毁释放资源
int pthread_cond_destroy(pthread_cond_t *cond);
参数:
- cond: 条件变量的地址
- attr: 条件变量属性, 一般使用默认属性, 指定为NULL
// 线程阻塞函数, 哪个线程调用这个函数, 哪个线程就会被阻塞
int pthread_cond_wait(pthread_cond_t *restrict cond, pthread_mutex_t *restrict mutex);
pthread_cond_wait
使线程等待条件变量的信号。当线程调用此函数时,它会原子性地释放与互斥锁关联的锁,并使线程休眠直到收到信号。
// 表示的时间是从1971.1.1到某个时间点的时间, 总长度使用秒/纳秒表示
struct timespec {
time_t tv_sec; /* Seconds */
long tv_nsec; /* Nanoseconds [0 .. 999999999] */
};
// 将线程阻塞一定的时间长度, 时间到达之后, 线程就解除阻塞了
int pthread_cond_timedwait(pthread_cond_t *restrict cond,
pthread_mutex_t *restrict mutex, const struct timespec *restrict abstime);
这个函数的前两个参数和pthread_cond_wait函数是一样的,第三个参数表示线程阻塞的时长,但是需要额外注意一点:struct timespec这个结构体中记录的时间是从1971.1.1到某个时间点的时间,总长度使用秒/纳秒表示。因此赋值方式相对要麻烦一点:
time_t mytim = time(NULL); // 1970.1.1 0:0:0 到当前的总秒数
struct timespec tmsp;
tmsp.tv_nsec = 0;
tmsp.tv_sec = time(NULL) + 100; // 线程阻塞100s
// 唤醒阻塞在条件变量上的线程, 至少有一个被解除阻塞
int pthread_cond_signal(pthread_cond_t *cond);
// 唤醒阻塞在条件变量上的线程, 被阻塞的线程全部解除阻塞
int pthread_cond_broadcast(pthread_cond_t *cond);
调用上面两个函数中的任意一个,都可以换线被pthread_cond_wait或者pthread_cond_timedwait阻塞的线程,区别就在于pthread_cond_signal是唤醒至少一个被阻塞的线程(总个数不定),pthread_cond_broadcast是唤醒所有被阻塞的线程。
7. 信号量
信号量是一种用于多线程或多进程之间同步和互斥的同步原语。它是一个计数器,用于控制同时访问共享资源的线程或进程的数量。信号量通常被用来解决竞争条件和协调多个线程或进程之间的操作。
信号量表示资源的数量,控制信号量的方式有两种原子操作:
- 一个是 P 操作,这个操作会把信号量减去 1,相减后如果信号量 < 0,则表明资源已被占用,进程需阻塞等待;相减后如果信号量 >= 0,则表明还有资源可使用,进程可正常继续执行。
- 另一个是 V 操作,这个操作会把信号量加上 1,相加后如果信号量 <= 0,则表明当前有阻塞中的进程,于是会将该进程唤醒运行;相加后如果信号量 > 0,则表明当前没有阻塞中的进程;
具体的过程如下:
- 进程 A 在访问共享内存前,先执行了 P 操作,由于信号量的初始值为 1,故在进程 A 执行 P 操作后信号量变为 0,表示共享资源可用,于是进程 A 就可以访问共享内存。
- 若此时,进程 B 也想访问共享内存,执行了 P 操作,结果信号量变为了 -1,这就意味着临界资源已被占用,因此进程 B 被阻塞。
- 直到进程 A 访问完共享内存,才会执行 V 操作,使得信号量恢复为 0,接着就会唤醒阻塞中的线程 B,使得进程 B 可以访问共享内存,最后完成共享内存的访问后,执行 V 操作,使信号量恢复到初始值 1。
可以发现,信号初始化为 1
,就代表着是互斥信号量,它可以保证共享内存在任何时刻只有一个进程在访问,这就很好的保护了共享内存。
信号量也可以用于消费者和生产者模型
另一种情况
具体过程:
- 如果进程 B 比进程 A 先执行了,那么执行到 P 操作时,由于信号量初始值为 0,故信号量会变为 -1,表示进程 A 还没生产数据,于是进程 B 就阻塞等待;
- 接着,当进程 A 生产完数据后,执行了 V 操作,就会使得信号量变为 0,于是就会唤醒阻塞在 P 操作的进程 B;
- 最后,进程 B 被唤醒后,意味着进程 A 已经生产了数据,于是进程 B 就可以正常读取数据了。
可以发现,信号初始化为 0
,就代表着是同步信号量,它可以**保证进程 A 应在进程 B 之前执行**。