掌握Linux线程编程的关键要点

什么是线程？

线程是CPU调度的基本单位，在Linux中是没有线程的概念的，取而代之的是轻量级进程，也叫做LWP。

线程和进程的区别与联系

进程由PCB和运行中的程序组成，PCB中有各种描述进程的字段，其中有一个指针struct mm_struct *mm指向虚拟地址空间。

进程在创建线程时，其实是创建了一个PCB，并且这个PCB里的mm指针是相同的。

红色的PCB就是用来描述创建出来的线程的。

CPU的调度单位是线程，所以进程创建出来多个线程后，这个“进程内部也就可以并发执行”。

线程拥有的资源有哪些？

线程独占的资源

• 线程的PCB（线程ID、线程状态、线程优先级等等）
• 寄存器里关于线程的上下文
• 线程的私有栈
• 信号屏蔽字
• errno
• 被__thread修饰的静态/全局字段，该字段虽然被创建在代码段，但是每个线程都会拥有自己的独立的实例，存放在私有栈中

线程共享的资源

• 运行中的程序
• 虚拟地址空间中的大部分（除掉栈和共享区中的私有栈）
• 文件描述符表
• 信号的处理方式
• 当前工作目录
• uid和gid

线程的优点和缺点分别是什么？

优点

• 创建线程时的代价要远小于进程
• 销毁线程时的代价要远小于进程
• 线程切换时的代价要远小于进程（其中cache中的数据更新的较少，而进程切换时cache中的数据会全部更新，要注意cache中的数据是热点数据）
• 线程能够共享虚拟地址空间，意味着线程间的数据共享要远比进程间的数据共享简单
• 合适数量的线程能够提高程序的运行效率

缺点

• 进程的健壮性（鲁棒性）降低，当一个线程发生段错误时就会导致OS发送信号进而中止掉整个进程（包括所有由该进程创建的线程）
• 线程由于共享虚拟地址空间，并且属于并发编程模式，所以会提高线程模型的编写难度以及出错概率
• 盲目的创建多个线程并不会提高程序的运行效率反而会降低，因为CPU会进行切片调度

线程的控制

线程的创建

#include 
int pthread_create(pthread_t *thread, const pthread_attr_t *attr,void *(*start_routine) (void *), void *arg);
1
2

通过pthread_create即可创建一个线程。

• 第一个参数是输出型参数，创建好的线程的标识符会写到这里
• 第二个参数是pthread_attr_t类型的指针，pthread_attr_t是一个结构体，可以供用户设置线程属性，一般很少使用
• 第三个参数是线程要执行的函数
• 第四个参数是传递给线程的参数

#include 
#include 
void *start_routine(void *arg) {
    std::cout << (char *)arg << std::endl;    
    return nullptr;
}
int main() {
    pthread_t tid;
    int n = pthread_create(&tid, nullptr, start_routine, (void *)"hello world");
    while(1);
    return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12

线程的标识符

当创建线程时，能够通过第一个输出型参数获取到线程标识符，也可以通过pthread_self函数获取调用者的线程标识符。

该标识符其实是描述一个线程单位的地址。

虚拟地址空间的共享区会加载原生线程库，而关于线程的一切操作都调用的是该库。当创建线程时，会在共享区中创建线程的TCB，创建多个线程时就会创建多个TCB，这些TCB被某种数据结构管理着。而线程的标识符就是TCB的入口地址。

线程的中止

线程可以通过return返回，还可以通过pthread_exit返回，但不可以通过exit或者是_exit返回。

因为调用exit或者是_exit，OS会向调用进程发送信号，导致整个进程退出，所以为了确保只是当前线程退出，应该使用return或pthread_exit。

#include 
void pthread_exit(void *retval);
1
2

return和pthread_exit在main函数和副线程函数中的不同表现

• 副线程return：副线程会正常退出，不会影响到别的线程
• 副线程pthread_exit：副线程会正常退出，不会影响到别的线程
• 主线程return：会导致整个进程退出
• 主线程pthread_exit：主线程退出，但变成僵尸线程（僵尸进程），不会影响其它线程，因为副线程有主线程回收资源，但主线程必须要父进程来回收资源，但主线程通过pthread_exit退出时，仅释放了主线程的PCB，整个进程还没退出，所以父进程就不回来回收资源，所以主线程pthread_exit时会变成僵尸线程。（强烈不推荐这样做，以上三种可以随意做）

线程的等待

进程在退出时，需要父进程来回收资源，线程在退出时，也需要创建它的线程来进行回收。

可以通过pthread_join进行等待。

#include 
int pthread_join(pthread_t thread, void **retval);
1
2

• 第一个参数是线程的标识符，表示我要等待这个线程
• 第二个参数是一个输出型参数，能够取到线程的返回值

#include                                                                                       
#include 
#include 
#include 
void *start_routine(void *arg) {
   int cnt = 3;
   while(cnt--) sleep(1);
   return (void *)"hello world!";
}
int main() {
   pthread_t tid;
   pthread_create(&tid, nullptr, start_routine, nullptr);
   void *retval = nullptr;
   pthread_join(tid, &retval);
   std::cout << (char *)retval << std::endl;
   return 0;                                
}            
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

为什么线程函数的返回值是一级指针，并且线程等待函数的第二个参数是二级指针呢？

因为这样副线程就能够返回任意的的数据（除了栈上的），比如副线程能够返回new出的一个类或者一个结构体的指针，而外部想要获取这个指针只能通过二级指针。

那么join函数是如何取到该指针的呢？

是因为线程函数的返回值回存放到pthread库中的一个地方，然后join函数去这个地方就能拿到该值。

线程的分离

当主线程不需要关心副线程的返回值的时候，那么主线程可以选择不join等待副线程，而是让副线程自己悄无声息的退出。

#include 
int pthread_detach(pthread_t thread);
1
2

该函数通过传入线程标识符，可以让该线程与主线程分离，进而当该线程退出时，不需要主线程去join。

该函数可以在主线程中使用pthread_detach(tid)，也可以在副线程中使用pthread_detach(pthread_self())

线程的取消

当副线程的任务不重要或者是运行到一半不需要了的时候，主线程可以手动中止该副线程。

 #include 
int pthread_cancel(pthread_t thread);
1
2

该函数传入要中止的线程标识符。

线程的高阶用法

临界资源的安全性如何保障？

多线程是共享虚拟地址空间的，所以多个线程间的数据共享是简单且自然的。最简单的就好比一个全局变量，都是多个线程共享的，那么多个线程同时访问该变量会不会出问题呢？

#include 
#include 
#include 
#include 

int gsize = 1000;

void *start_routine(void *arg) {
    while (gsize > 0) {
        usleep(1000);
        --gsize;
    }
    return nullptr;
}

int main() {
    pthread_t tid;
    pthread_create(&tid, nullptr, start_routine, nullptr);
    while (gsize > 0) {
        usleep(1000);
        --gsize;
    }
    pthread_join(tid, nullptr);
    std::cout << gsize << std::endl;
    return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

很明显，两个线程的逻辑分别都是当gsize>0时就- -，当gsize == 0时就退出。所以最终的答案应该是0才对，而这里的答案是-1。

那是因为当两个线程同时访问gsize的时候，假设此时gsize为1，而两个线程while判断的时候都是1 > 0，从而都减了两次，所以答案会是-1。

usleep的目的是，加速cpu切片的频率（因为当内核态向用户态进行转换的时候会检测是否需要切片，当合适的时候就切），以便更容易出现实验效果。

多个线程都能访问到的资源叫做临界资源，上面的gsize就是临界资源，而访问临界资源的代码就是临界区，代码中的两个while块。

如何保证临界资源的安全性呢？可以通过互斥锁、信号量等。

互斥锁

要保证临界资源的安全性，只需保证以下几点：

• 当多个线程要进入临界区时，只能允许一个进去
• 当临界区中有人时，别人不能进去
• 一个线程不能妨碍别的线程进入临界区

那么要做到这几点，只需要一把互斥锁，在多个线程要进去临界区时就去抢这把锁，谁抢到谁就进，进入临界区的人把门锁住，别人也就进不去了，当临界区里的人出来时，再把门打开，把锁放回去，其它线程再去抢锁。

定义互斥锁

互斥锁就是一个pthread_mutex_t类型的变量。

定义一个名为mtx的互斥锁：pthread_mutex_t mtx

初始化互斥锁

int pthread_mutex_init(pthread_mutex_t *restrict mutex,const pthread_mutexattr_t *restrict attr);
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
1
2

局部变量的互斥锁，通过上面的函数进行初始化。

全局变量的互斥锁，通过上面的宏进行初始化。

• 初始化函数中的第一个参数是锁的指针
• 初始化函数中的第二个参数是设置锁的属性，一般情况下都传入NULL

销毁互斥锁

int pthread_mutex_destroy(pthread_mutex_t *mutex);
1

传入锁的指针

上锁与解锁

int pthread_mutex_lock(pthread_mutex_t *mutex);
int pthread_mutex_trylock(pthread_mutex_t *mutex);
int pthread_mutex_unlock(pthread_mutex_t *mutex);
1
2
3

• 第一个函数用来获取锁，当获取成功时返回0，获取失败时阻塞
• 第二个函数也用来获取锁，当获取成功时返回0，获取失败时不会阻塞，而是返回一个错误码
• 第三个函数用来释放锁

使用互斥锁将上面代码修正

#include 
#include 
#include 
#include 

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

int gsize = 1000;

void *start_routine(void *arg) {
    while (true) {
        pthread_mutex_lock(&mutex);
        if (gsize > 0) {
            usleep(1000);
            --gsize;
            pthread_mutex_unlock(&mutex);
        } else {
            pthread_mutex_unlock(&mutex);
            break;
        }
    }
    return nullptr;
}

int main() {
    pthread_t tid;
    pthread_create(&tid, nullptr, start_routine, nullptr);
    while (true) {
        pthread_mutex_lock(&mutex);
        if (gsize > 0) {
            usleep(1000);
            --gsize;
            pthread_mutex_unlock(&mutex);
        } else {
            pthread_mutex_unlock(&mutex);
            break;
        }
    }
    pthread_join(tid, nullptr);
    std::cout << gsize << std::endl;
    pthread_mutex_destroy(&mutex);
    return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

互斥锁的原理

其实互斥锁本身也是一个临界资源，那么互斥锁的安全是如何保证的呢？

因为互斥锁的上锁与解锁是原子性的。

// 伪代码
lock:
			  movb $0, %al
        xchgb %al, mutex
        if (al寄存器的内容 > 0) {
            return 0;
        } else {
          	阻塞挂起
        }
     	  被操作系统唤醒后
			  goto lock
unlock:
			  movb $1, mutex
        唤醒等待mutex的线程
        return 0
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

互斥锁导致的线程饥饿问题

先看一下现象

#include 
#include 
#include 
#include 

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

int gsize = 1000;

void *start_routine(void *arg) {
    while (true) {
        pthread_mutex_lock(&mutex);
        if (gsize > 0) {
            usleep(1000);
          	std::cout << (char *)arg << "抢到锁了" << std::endl;
            --gsize;
            pthread_mutex_unlock(&mutex);
        } else {
            pthread_mutex_unlock(&mutex);
            break;
        }
    }
    return nullptr;
}

int main() {
    pthread_t tid;
    pthread_create(&tid, nullptr, start_routine, nullptr);
    while (true) {
        pthread_mutex_lock(&mutex);
        if (gsize > 0) {
            usleep(1000);
          	std::cout << (char *)arg << "抢到锁了" << std::endl;
            --gsize;
            pthread_mutex_unlock(&mutex);
        } else {
            pthread_mutex_unlock(&mutex);
            break;
        }
    }
    pthread_join(tid, nullptr);
    std::cout << gsize << std::endl;
    pthread_mutex_destroy(&mutex);
    return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

可以发现，我们虽然是两个线程在共同竞争同一把锁，但总是副线程能抢到，而主线程抢不到。

这是因为，当多个线程因为竞争锁而被阻塞时，被唤醒的顺序是不确定的，当副线程释放锁时，副线程也就有更大的概率能抢上锁（因为锁总是在副线程手中，副线程刚释放，那么在众多竞争锁的线程中，副线程是距离这把锁最近的线程）。那么这就会导致其它线程总是在竞争锁而竞争不到从而一直在阻塞。这就是互斥锁导致的线程饥饿问题。

那么要解决这一问题，就必须让竞争锁的线程按照一定的顺序进行排队。那么就出现了条件变量。

条件变量

条件变量能够让因为申请互斥锁而导致阻塞的线程按照申请锁的顺序进行排队，这些线程被唤醒时也是按照这个排队顺序被唤醒的。也就能够解决上面的线程饥饿问题，因为当释放锁的线程再次去申请锁时，必须先申请条件变量。

定义条件变量

条件变量就是一个pthread_cond_t类型的变量。

定义一个名为cond的条件变量：pthread_cond_t cond

初始化条件变量

int pthread_cond_init(pthread_cond_t *restrict cond,const pthread_condattr_t *restrict attr);
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
1
2

局部变量的条件变量，通过上面的函数进行初始化。

全局变量的条件变量，通过上面的宏进行初始化。

• 初始化函数中的第一个参数是条件变量的指针
• 初始化函数中的第二个参数是设置条件变量的属性，一般情况下都传入NULL

销毁条件变量

int pthread_cond_destroy(pthread_cond_t *cond);
1

申请条件变量

#include 
int pthread_cond_timedwait(pthread_cond_t *restrict cond,pthread_mutex_t *restrict mutex,const struct timespec *restrict abstime);
int pthread_cond_wait(pthread_cond_t *restrict cond,pthread_mutex_t *restrict mutex);
1
2
3

• wait函数的第一个参数是条件变量的指针，第二个参数是已申请了锁的指针，当调用wait时，线程会被阻塞挂起，当有别的线程唤醒该条件变量上的线程时，wait会返回0。
• timedwait函数的第三个参数设置了一个定时器，当时间到了时，该函数会返回TIMEOUT

唤醒条件变量

#include 
int pthread_cond_broadcast(pthread_cond_t *cond);
int pthread_cond_signal(pthread_cond_t *cond);
1
2
3

• signal只会唤醒cond上的第一个线程
• broadcast会唤醒cond上的所有线程

使用条件变量量修改上面代码

#include 
#include 
#include 
#include 

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;

int gsize = 1000;

void *start_routine(void *arg) {
    while (true) {
        pthread_mutex_lock(&mutex);
        pthread_cond_wait(&cond, &mutex);
        if (gsize > 0) {
            usleep(1000);
            std::cout << (char *)arg << "抢到锁了" << std::endl;
            --gsize;
            pthread_mutex_unlock(&mutex);
        } else {
            pthread_mutex_unlock(&mutex);
            break;
        }
    }
    return nullptr;
}

int main() {
    pthread_t tid1, tid2;
    pthread_create(&tid1, nullptr, start_routine, (void *)"副线程1号");
    pthread_create(&tid2, nullptr, start_routine, (void *)"副线程2号");
    while (true) {
        pthread_cond_signal(&cond);
        sleep(1);
    } 
    pthread_join(tid1, nullptr);
    pthread_join(tid2, nullptr);
    pthread_mutex_destroy(&mutex);
    pthread_cond_destroy(&cond);
    return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

条件变量的原理

一个条件变量对应一个等待对列，每个调用wait函数的线程都会在这个队列上。

当调用signal或broadcast时，会唤醒这个条件变量对应的队列上的线程。

信号量

像单纯的互斥锁，只能保证一个整体的互斥，但有时，我们并不是要一个整体互斥，比如电影院买票，一个大厅有100个座位，我们用互斥锁对整个大厅进行封锁是不合适的，这时就可以用信号量了。

定义信号量

信号量是一个sem_t类型的变量，定义一个名为sem的信号量：sem_t sem

初始化信号量

 #include 
int sem_init(sem_t *sem, int pshared, unsigned int value);
1
2

• 第一个参数传入信号量的指针
• 第二个参数为0，意味着这个信号量是一个进程内部的线程共享，若为非0，意味着这个信号量是多个进程间共享
• 第三个参数指定这个信号量的初始值是多少

销毁信号量

#include 
int sem_destroy(sem_t *sem);
1
2

申请资源

#include 
int sem_wait(sem_t *sem);
int sem_trywait(sem_t *sem);
int sem_timedwait(sem_t *sem, const struct timespec *abs_timeout);
1
2
3
4

该类函数用于申请资源，每次调用这个函数，都会使得信号量值减1，若当信号量为0时还调用该函数则会阻塞挂起。

• 第一个函数就是简单的申请资源，资源-1，若资源为0就会阻塞挂起
• 第二个函数不会阻塞挂起，而是在资源为0的时候返回EAGAIN
• 第三个函数若是在资源为0时会阻塞abs_timeout时间，等时间到时会返回-1并且errno设置ETIMEDOUT

该操作也经常被称为P操作

释放资源

#include 
int sem_post(sem_t *sem);
1
2

信号量加1，若信号量大于0了，则会唤醒阻塞的线程。

该操作也经常被称为V操作

信号量的原理

wait(s) {
  	while (s <= 0) {
      	阻塞挂起
    }
  	--s
}

post(s) {
  	++s
}
1
2
3
4
5
6
7
8
9
10

什么是死锁？如何避免死锁？

A线程
lock(mutex1)
lock(mutex2)
// 业务逻辑
unlock(mutex2)
unlock(mutex1)
  
B线程
lock(mutex2)
lock(mutex1)
// 业务逻辑
unlock(mutex1)
unlock(mutex2)
1
2
3
4
5
6
7
8
9
10
11
12
13

若A线程在获取了1锁后被切走了，此时B线程又获取了2锁，那么此时就形成了死锁。A线程想继续获取2锁但因为B线程已经获取了所以A线程被挂起，B线程想继续获取1锁但因为A线程已经获取了所以B线程也被挂起，那么至此A、B两个线程都被挂起，除非有外界干预，否则这两个线程将一直保持挂起状态。

死锁的四个必要条件

• 首先要有线程访问临界资源时互斥
• 当有线程已经获取了锁时，它还想获取锁
• 当有线程已经获取了锁时，别的线程不能把锁给剥夺过来
• 若干执行流要形成一种循环等待的状态，就如同上面例子

如何避免死锁

• 破坏上面四个必要条件之一。
• 多个锁的加锁、解锁顺序一致
• 避免锁未被释放
• 多个锁可能意味着多个临界资源，想办法将这多个临界资源一次性获取即用一把锁

生产者消费者模型

简单理解

生活中处处存在生产者消费者模型的例子

这就是生活中最常见最典型的例子了。

在日常生活中，我们也很少会直接去供货商那买东西，因为麻烦，为了方便所以我们总是去超市买东西。供货商也不会直接在马路上拉个人就卖给他，因为这样卖出去的量太少了。所以，在日常生活中，若是生产者和消费者直接接触，那么经济交易的效率就会大大降低。所以就诞生了超市这个角色，供货商把大量的产品批发给超市，而不用担心自己的产品滞销，老百姓之间去超市买东西，既简单便捷，而且能够买到产品的种类也很多。这就使得生产者和消费者之间的联系变得非常微弱。

在计算机中也需要这样的生产者消费者模型

发布任务的线程就如同生产者，执行任务的线程就如同消费者。若是这两种线程直接接触，那么效率就会大大降低，若是在这两种线程之间多一个缓冲区，这个缓冲区就如同超市，生产者把发布的任务放缓冲区里，消费者去缓冲区里拿要执行的任务，这样就能够实现两种线程之间的解藕。

生产者消费者模型的优点

• 实现了生产者和消费者之间的解藕
• 支持生产者和消费者之间的并发
• 生产者和消费者可以是忙闲不均的

基于条件变量的阻塞队列类型的生产者消费者模型

#include 
#include 
#include 

const static size_t gcap = 10;

template
class BlockQueue {
public:
    BlockQueue() : cap_(gcap) {
        pthread_mutex_init(&mutex_, nullptr);
        pthread_cond_init(&producer_cond_, nullptr);
        pthread_cond_init(&consumer_cond_, nullptr);
    } 
    ~BlockQueue() {
        pthread_mutex_destroy(&mutex_);
        pthread_cond_destroy(&producer_cond_);
        pthread_cond_destroy(&producer_cond_);
    }
    void push(const T &in) {
        pthread_mutex_lock(&mutex_);
        while (is_full()) {
            pthread_cond_wait(&producer_cond_, &mutex_);
        }        
        bq_.push(in);
        pthread_cond_signal(&consumer_cond_);
        pthread_mutex_unlock(&mutex_);
    }
    void pop(T *out) {
        pthread_mutex_lock(&mutex_);
        while (is_empty()) {
            pthread_cond_wait(&consumer_cond_, &mutex_);
        }
        *out = bq_.front();
        bq_.pop();
        pthread_cond_signal(&producer_cond_);
        pthread_mutex_unlock(&mutex_);
    }
private:
    bool is_empty() { return bq_.size() == 0; }
    bool is_full() { return bq_.size() == cap_; }
private:
    std::queue bq_;
    size_t cap_; 
    pthread_mutex_t mutex_; 
    pthread_cond_t producer_cond_;
    pthread_cond_t consumer_cond_; 
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

基于信号量的循环队列类型的生产者消费者模型

#include 
#include 
#include 
#include 

const static size_t gcap = 10;

template
class RingQueue {
public:
    RingQueue() : cap_(cap), space_step_(0), data_step_(0) {
        rq_.resize(cap_);
        pthread_mutex_init(&mutex_, nullptr);
        sem_init(space_sem_, 0, cap_);
        sem_init(data_sem_, 0, 0);
        pthread_mutex_init(&producer_mutex_, nullptr);
        pthread_mutex_init(&consumer_mutex_, nullptr);
    }
    ~RingQueue() {
        pthread_mutex_destroy(&mutex_);
        sem_destroy(&space_sem_);
        sem_destroy(&data_sem_);
        pthread_mutex_destroy(&producer_mutex_);
        pthread_mutex_destroy(&consumer_mutex_);
    }
    void push(const T& in) {
        P(&space_sem_);
        pthread_mutex_lock(&producer_mutex_);
        rq_[space_step_++] = in;
        space_step_ %= cap_;
        pthread_mutex_unlock(&producer_mutex_);
        V(&data_sem_); 
    }
    void pop(T *out) {
        P(&data_sem_);
        pthread_mutex_lock(&consumer_mutex_);
        *out = rq_[data_step_++];
        data_step_ %= cap_;
        pthread_mutex_unlock(&consumer_mutex_);
        V(&space_sem_);
    }
private:
    void P(sem_t *sem) { sem_wait(sem); }
    void V(sem_t *sem) { sem_post(sem); }
private: 
    std::vector rq_;
    size_t cap_;
    pthread_mutex_t mutex_;
    sem_t space_sem_;
    sem_t data_sem_;
    int space_step_;
    int data_step_;
    pthread_mutex_t producer_mutex_;
    pthread_mutex_t consumer_mutex_;
}; 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55