十七、多线程（下）_线程时序

一、线程互斥，它是对的，但是不合理（饥饿问题）——同步

不合理：互斥有可能导致饥饿问题——由于执行流1优先级高，她就不断的申请锁，释放锁，则另一个执行流2会长时间得不到某种资源。
在保证临界资源安全的前提下(互斥等)，让线程访问某种资源，具有一定的顺序性，称之为同步。

同步：在保证数据安全的前提下，让线程能够按照某种特定的顺序访问临界资源，从而有效避免饥饿问题，叫做同步。
竞态条件：因为时序问题，而导致程序异常，我们称之为竞态条件。在线程场景下，这种问题也不难理解同步能够解决线程互斥不合理性的问题：防止饥饿，线程协同。

二、条件变量

（一）概念

当一个线程互斥地访问某个变量时，它可能发现在其它线程改变状态之前，它什么也做不了。
例如一个线程访问队列时，发现队列为空，它只能等待，直到其它线程将一个节点添加到队列中。这种情况就需要用到条件变量。
条件(对应的共享资源的状态，程序员要判断资源是否满足自己操作的要求，为满/为空就不满足)，条件变量(条件满足或者不满足的时候，进行wait或signal–种方式)。

条件变量：通过判断条件是否满足要求来决定是否让当前线程等待。

（二）条件变量接口

1. pthread_cond_init 创建条件变量

• pthread_cond_t cond = PTHREAD_COND_INITIALIZER; 定义全局/静态的条件变量，可以用这个宏初始化

• int pthread_cond_init(pthread_cond_t *restrict cond,const pthread_condattr_t *restrict attr); 定义局部条件变量—— cond：条件变量的地址。attr：条件变量的属性设为空

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

2. pthread_cond_wait 等待条件满足

让对应的线程进行等待，等待被唤醒，即调用这个接口线程会被阻塞。

条件变量要和mutex互斥锁，一并使用，为什么？

条件变量的wait中需要传入锁的意义是：在阻塞线程的时候，会自动释放mutex_锁。

int pthread_cond_wait(pthread_cond_t *restrict cond,pthread_mutex_t *restrict mutex);
参数：
cond：要在这个条件变量上等待
mutex：互斥量，后面详细解释
1
2
3
4

3. pthread_cond_destroy 销毁条件变量

int pthread_cond_destroy(pthread_cond_t *cond)
1

4. 唤醒等待

int pthread_cond_broadcast(pthread_cond_t *cond);  //唤醒一个在指定条件变量下等待的所有线程
int pthread_cond_signal(pthread_cond_t *cond);  //唤醒一个在指定条件变量下等待的线程，一个一个唤醒时，所有线程以队列方式排列的

1
2
3

（三）例子

#include<vector>
#include<iostream>
#include<pthread.h>
#include<functional>

#include<unistd.h>
using namespace std;

pthread_cond_t cond;

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

vector<function<void()>> funcs;

void show() {
    cout<<"hello show"<<" thread:"<<pthread_self()<<endl;
}
void print() {
    cout << "hello print" << endl;
}
void* waitCommand(void* args) {
    pthread_detach(pthread_self());
    //cout<<"!!!!!!!111111"<<endl;
    while(true)
    {
        pthread_cond_wait(&cond,&mutex);
        for(auto& f:funcs)
        {
            f();
        }
    }
    cout<<"thread id: "<<pthread_self()<<" end... "<<endl;
    return nullptr;

}
int main() {
    funcs.push_back(show);
    funcs.push_back(print);
    funcs.push_back([](){
        cout<<"你好，条件变量！"<<endl;
    });
    pthread_cond_init(&cond,nullptr);

    pthread_t t1,t2,t3;
    pthread_create(&t1,nullptr,waitCommand,nullptr);
    pthread_create(&t2,nullptr,waitCommand,nullptr);
    pthread_create(&t3,nullptr,waitCommand,nullptr);

    while(true) {
        sleep(1);
        pthread_cond_broadcast(&cond);
    }
    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
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

三、生产者消费者模型

生产者消费者模型——同步与互斥的最典型的应用场景——重新认识条件变量

（一）321模型

• 生产者和生产者(互斥)，消费者和消费者(互斥)，生产者和消费者( 互斥/同步)——3种关系
• 生产者和消费者——由线程承担的 2种角色
• 超市:内存中特定的一种内存结构(数据结构)——1个交易场所
  
  这里的仓库就是缓冲区，也是临界资源：①提高效率。②解耦生产者和消费者之间的耦合关系。
  （一般就是内存中的一段空间，可以有自己的组织方式）
• 消费者有多个，消费者之间是竞争关系——互斥关系
• 生产者有多个，生产者之间也是竞争关系——互斥关系
• 消费者和生产者之间又是什么关系呢?
  1.互斥关系：生产者把"123456789"写入缓冲区时（正在生产），消费者突然来拿，可能只拿走了"12345"就错误了，所以消费者和生产者之间也要有互斥关系。
  2.同步关系：要有一定的顺序去执行——消费完了要生产，生产满了要消费

（二）为何要使用生产者消费者模型

生产者消费者模式就是通过一个容器来解决生产者和消费者的强耦合问题。
生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。这个阻塞队列就是用来给生产者和消费者解耦的。

（三）生产者消费者模型优点

• 解耦
• 支持并发
• 支持忙闲不均

（四）基于BlockingQueue的生产者消费者模型

1. 阻塞队列概念

在多线程编程中阻塞队列 (Blocking Queue) 是一种常用于实现生产者和消费者模型的数据结构。
其与普通的队列区别在于:

• 当队列为空时，从队列获取元素的操作将会被阻塞，直到队列中被放入了元素；
• 当队列满时，往队列里存放元素的操作也会被阻塞，直到有元素被从队列中取出( 以上的操作都是基于不同的线程来说的，线程在对阻塞队列进程操作时会被阻塞)，管道就是一种阻塞队列。
  

2. BlockingQueue阻塞队列代码

（1）pthread_cond_wait(&conCond_, &mutex_); 解锁

挂起时生产者一定是在临界区中的，因为你要先上锁，不上锁就没有资格访问临界资源（如果不上锁就访问临界资源，则不符合同步的规则，那就是代码写的有问题）。此时上锁后被挂起，生产者和消费者用的同一把锁，如果 生产者/消费者 不释放锁（解锁）那么对方的线程就永远无法访问，所以条件变量的wait中需要传入锁的意义是：在阻塞线程的时候，会自动释放mutex_锁（解锁）

（2）生产者push时用while循环判断

while (isFull())：原因是 proBlockWait();调用的 pthread_cond_wait(&proCond_, &mutex_); 有可能返回失败，或被伪唤醒（伪唤醒可能是系统造成的或者写的代码有错误造成）。

如果是 if (isFull()) 则返回失败/伪唤醒后 就直接向下指向条件满足的代码，可是此时阻塞队列还是满的，再添加数据就错了；

所以用while (isFull()) 返回失败/伪唤醒后需要继续判断是否为满，为满就是返回失败/伪唤醒导致的—>重新等待；不满就是等待成功，消费者也消费了—>就可以添加数据了

三、POSIX信号量

一份公共资源，但是允许同时访问不同的区域！

不同的线程并发访问公共资源不同的区域！

（一）概念

POSIX信号量和 SystemV 信号量作用相同，都是用于同步操作，达到无冲突的访问共享资源目的， 但 POSIX 可以用于线程间同步。

信号量：是一个计数器!
只要拥有信号量，就在未来一定能够拥有临界资源的一部分！

申请信号量的本质：对临界资源中特定小块资源的 <预定> 机制。

例子： 就和你在电影院买票一样，
你买了票，等于你预定了座位！！！
如果你没有买票，就进去坐着，这样是不符合规定的！

只要申请成功，就一定有你得资源！
只要申请失败，就说明条件不就绪，你只能等！

信号量的PV操作：V ++归还资源，P --申请资源。信号量的作用：限制进入临界区的线程个数。

线程要访问临界资源中的某一块区域 ——> 申请信号量 —— > 得先看到信号量 ——> 信号量本身必须是：公共资源

信号量的PV操作：
sem – ; 申请资源 —— 必须保证操作的原子性 p

sem ++ ; 归还资源 ——必须保证操作的原子性 v

（二）信号量接口

1. sem_init 初始化一个未命名的信号量

man 3 sem_init

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

sem：初始化的信号量。
pshared：0 表示线程间共享，非零表示进程间共享（填0）。
value ：信号量初始值

2. sem_destroy 销毁一个信号量

int sem_destroy(sem_t *sem);
1

3. sem_wait 使用信号量（占座）

功能：等待信号量，会将信号量的值减1

int sem_wait(sem_t *sem); // P --;
1

4. sem_post 归还信号量（退座）

功能：发布信号量，表示资源使用完毕，可以归还资源了，将信号量值加1 。

int sem_post(sem_t *sem); // V ++
1

上面生产者 - 消费者的例子是基于 queue 的 , 其空间可以动态分配 , 现在基于固定大小的环形队列重写这个程序 （POSIX 信号量）。

三、环形队列

（一）信号量的作用

后续操作基本原则:（信号量保证满数据情况只能是消费线程先消费数据资源，数据为空的情况下生产线程先申请空间资源）。

环形队列有可能访问同一个位置。什么时候会发生?

我们两个指向同一个位置的时候只有满or空的时候! ( 互斥and同步)其他时候，都指向不同的位置! (并发 )。

1.数据为空：消费者不能超过生产者一>生产者先运行

生产者:最关心的是什么资源：空间默认是N: [N, 0]

2.数据为满：生产者不能把消费者套一个圈然后继续再往后写入——消费者先运行

消费者:最关心的是什么资源：数据默认是0 [0，N]

生产线程生产，P(roomSem)申请一个空间资源 --、V(dataSem)释放一个数据资源++；

消费线程消费，P(dataSem)申请一个数据资源 --、V(roomSem)释放一个空间资源++；

• 哪个线程先运行不能保证，但是数据为空的情况下能保证生产线程先申请空间资源，因为生产者关心的空间资源计数器（信号量）默认是N，消费者关心的数据资源计数器（信号量）默认是0，即使消费者线程先运行，因为数据资源计数器默认是0，无法再P–，消费者也无法消费数据资源，消费者会被挂起。
• 只要生产者生产完之后消费者才能消费，满数据的情况只能是消费线程先消费数据资源，同理。
  我们是单生产者，单消费者

多生产者，多消费者，代码怎么改？
为什么呢？？？多生产者，多消费者？
不要只关心把数据或者任务，从ringqueue 放拿的过程，获取数据或者任务，处理数据或者任务，也是需要花时间的！，而这个过程是多线程并发进行的，因为上述工作的准备也需要时间！

（二）code

1. 单生产者，单消费者

具体细节可以看我的这篇博客！
http://t.csdn.cn/Zlqsm

2.多生产者，多消费者

与单线程的区别：要上锁，使生产者与生产者之间生产互斥；消费者与消费者之间消费互斥。
如果 sem_wait(&roomSem_);P–操作放在lock后面，sem_post(&dataSem_); 放在unlock前面，每次只能进入一个线程申请信号量，则信号量无法被多次的申请，只能互斥申请。

但信号量本身就是资源的预定机制，就应该让多个资源一起申请，然后再一个一个排队执行锁内的生产过程。
所以我们正确的顺序应该是：
先申请信号量，然后上锁！
申请信号量是原子的，生产过程是非原子的，所以生产过程要被保护，消费过程同理。

具体细节可以看我的这篇博客！
http://t.csdn.cn/Zlqsm

四、锁

（一）STL，智能指针和线程安全

1.STL中的容器是否是线程安全的?

不是。
原因是 , STL 的设计初衷是将性能挖掘到极致 , 而一旦涉及到加锁保证线程安全 , 会对性能造成巨大的影响 。
方式的不同 , 性能可能也不同 ( 例如 hash 表的锁表和锁桶 )。
因此 STL 默认不是线程安全 . 如果需要在多线程环境下使用 , 往往需要调用者自行保证线程安全 。

2.智能指针是否是线程安全的?

对于 unique_ptr, 由于只是在当前代码块范围内生效 , 因此不涉及线程安全问题。
对于 shared_ptr, 多个对象需要共用一个引用计数变量 , 所以会存在线程安全问题。但是标准库实现的时候考虑到了这个问题, 基于原子操作 (CAS) 的方式保证 shared_ptr 能够高效 , 原子的操作引用计数。（ shared_ptr是线程安全的 ）。

3.其他常见的各种锁？

悲观锁：在每次取数据时，总是担心数据会被其他线程修改，所以会在取数据前先加锁（读锁，写锁，行锁等），当其他线程想要访问数据时，被阻塞挂起。
乐观锁：每次取数据时候，总是乐观的认为数据不会被其他线程修改，因此不上锁。但是在更新数据前，会判断其他数据在更新前有没有对数据进行修改。主要采用两种方式：版本号机制和CAS 操作。
CAS 操作：当需要更新数据时，判断当前内存值和之前取得的值是否相等。如果相等则用新值更新。若不等则失败，失败则重试，一般是一个自旋的过程，即不断重试。
自旋锁（ phtread_spin_init/lock ），公平锁，非公平锁。

故事：你请张三吃饭，但是张三说需要再等1个小时，那么你会去上网（挂起），然后等会再来找他；如果张三说需要再等一小会儿，那么你会在楼下等，并且等一会儿就给他打个电话问问好了没，张三说马上，你就再等一会儿，然后再打电话。

是什么决定了你在楼下，是以什么方式等待张三的? ?
在临界区中等待时间决定的
①等待时间短——>轮询测试，是否就绪。pthread_ spin_ lock（phtread_spin_init/lock，使用自旋锁只需要mutex->spin即可
PCB—>S —>阻塞的；pthread—>while(true)—>阻塞的!
②等待时间长——>去上网的路上，上网（挂起），回来的路上->阻塞前，阻塞中，阻塞后。挂起等待锁（在临界区中IO操作，系统调用花时间就长）

概念：本质是对执行流的预先分配，当有任务时直接指定线程完成对应任务，而不需要再现场创建线程。

五、线程池

加粗样式## （一）概念
线程池：对执行流的预先分配，当有任务时会直接指定线程去做，而不需要再创建线程。

（二）特点

①多线程程序的运行效率， 是一个正态分布的结果， 线程数量从1开始增加， 随着线程数量的增加， 程序的运行效率逐渐变高， 直到线程数量达到一个临界值， 当在增加线程数量时， 程序的运行效率会减小（主要是由于频繁线程切换影响线程运行效率）

②降低资源消耗：通过重用已经创建的线程来降低线程创建和销毁的消耗（解释：线程池中更多是对已经创建的线程循环利用，因此节省了新的线程的创建与销毁的时间成本）

③提高线程的可管理性：线程池可以统一管理、分配、调优和监控（解释：线程池是一个模块化的处理思想，具有统一管理，资源分配，调整优化，监控的优点）

④降低程序的耦合程度： 提高程序的运行效率（线程池模块与任务的产生分离，可以动态的根据性能及任务数量调整线程的数量，提高程序的运行效率）

（三）线程池的实现原理与作用

1.实现原理：

线程池通过一个线程安全的阻塞任务队列加上一个或一个以上的线程实现，线程池中的线程可以从阻塞队列中获取任务进行任务处理，当线程都处于繁忙状态时可以将任务加入阻塞队列中，等到其它的线程空闲后进行处理。

2.作用：

可以避免大量线程频繁创建或销毁所带来的时间成本，也可以避免在峰值压力下，系统资源耗尽的风险；并且可以统一对线程池中的线程进行管理，调度监控。

线程池 :
一种线程使用模式。线程过多会带来调度开销，进而影响缓存局部性和整体性能。而线程池维护着多个线程，等待着
监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能够保证内核的充分利
用，还能防止过分调度。可用线程数量应该取决于可用的并发处理器、处理器内核、内存、网络 sockets 等的数量。

• 线程池的应用场景：

1. 需要大量的线程来完成任务，且完成任务的时间比较短。 WEB 服务器完成网页请求这样的任务，使用线程池技
   术是非常合适的。因为单个任务小，而任务数量巨大，你可以想象一个热门网站的点击次数。 但对于长时间的任务，比如一个
   Telnet 连接请求，线程池的优点就不明显了。因为 Telnet 会话时间比线程的创建时间大多了。
2. 对性能要求苛刻的应用，比如要求服务器迅速响应客户请求。
3. 接受突发性的大量请求，但不至于使服务器因此产生大量线程的应用。突发性大量客户请求，在没有线程池情
   况下，将产生大量线程，虽然理论上大部分操作系统线程数目最大值不是问题，短时间内产生大量线程可能使内存到达极限，
   出现错误 .

线程池的种类：
线程池示例：

• 创建固定数量线程池，循环从任务队列中获取任务对象。
• 获取到任务对象后，执行任务对象中的任务接口。

易错点：

① start()中assert忘写
②线程池用智能指针维护
③因为类内的函数threadRoutine多传了一个this指针，所以和pthread_ create(&temp, nullptr, threadRoutine,this);里面的threadRoutine类型不匹配了，所以threadRoutine函数定义成static去掉隐含this才能类型匹配，并且pthread_create传入this指针。
④choiceThreadForHandler(); 选择一个线程去执行

void push(const T &in)
    {
        LockGuard lockguard(&_mutex);
        _task_queue.push(in);
        pthread_cond_signal(&_cond);
    }
1
2
3
4
5
6

(1)Task.hpp

#pragma once

#include <iostream>
#include <string>
#include <functional>
#include <cstdio>

class Task {
     using func_t = function<int(int,int,char)>;
public:
    Task() {

    }
    Task(int x, int y, char op, func_t func)
    :_x(x), _y(y), _op(op), _callback(func)
    {}
    string operator() () {
        int result = _callback(_x,_y,_op);
        char buffer[1024];
        snprintf(buffer,sizeof buffer,"%d %c %d = %d",_x,_op,_y,result);
        return buffer;
    }
    string toTaskString() {
        char buffer[1024];
        snprintf(buffer,sizeof buffer,"%d %c %d = ?", _x, _op, _y);
        return buffer;
    }
private:
    int _x;
    int _y;
    char _op;
    func_t _callback;
};

const std::string oper = "+-*/%";
int mymath(int x, int y, char op)
{
    int result = 0;
    switch (op)
    {
    case '+':
        result = x + y;
        break;
    case '-':
        result = x - y;
        break;
    case '*':
        result = x * y;
        break;
    case '/':
    {
        if (y == 0)
        {
            std::cerr << "div zero error!" << std::endl;
            result = -1;
        }
        else
            result = x / y;
    }
        break;
    case '%':
    {
        if (y == 0)
        {
            std::cerr << "mod zero error!" << std::endl;
            result = -1;
        }
        else
            result = x % y;
    }
        break;
    default:
        // do nothing
        break;
    }

    return result;
}
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
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

(2)LockGuard.hpp

#pragma once

#include<iostream>
#include<pthread.h>

class Mutex {
public:
    Mutex(pthread_mutex_t *lock_p = nullptr) : lock_p_(lock_p) {

    }
    void lock() {
        if (lock_p_) pthread_mutex_lock(lock_p_);
    }
    void unlock() {
        if (lock_p_) pthread_mutex_unlock(lock_p_);
    }
    ~Mutex()
    {}
private:
    pthread_mutex_t *lock_p_;
};

class LockGuard {
public:
    LockGuard(pthread_mutex_t *mutex) : mutex_(mutex) {
        mutex_.lock();
    }
    ~LockGuard()
    {
        mutex_.unlock(); //在析构函数中进行解锁
    }
private:
    Mutex 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

(3)Thread.hpp

#pragma once

#include<iostream>
#include<unistd.h>
#include<cstdio>
#include<pthread.h>
#include<cstring>
#include<string>
#include<functional>
using namespace std;

class Thread;
// 上下文,当成一个大号的结构体
// class Context {
// public:
//     Thread *this_;
//     void *args_;
// public:
//     Context():this_(nullptr),args_(nullptr) {

//     }
//     ~Context()
//     {}
// };
namespace ThreadNs {
    typedef function<void*(void *)> func_t;
    const int num = 1024;

    class Thread {
    private:
      static void *start_routine(void *args) {  //类内成员，有缺省参数！
            Thread *_this  = static_cast<Thread *> (args);
            return _this->callback();
            // void * ret = ctx -> this_ -> run(ctx -> args_);
            // delete ctx;
            // return ret;
            // 静态方法不能调用成员方法或者成员变量
        }
    public:
        Thread(func_t func,void * args = nullptr):func_(func),args_(args) {
            char buffer[num];

            snprintf(buffer,sizeof buffer,"thread-%d",threadnum++);
            name_ = buffer;

            // Context *ctx = new Context();

            // ctx -> this_ = this;
            // ctx -> args_ = args;

          

        }
        //  // 在类内创建线程，想让线程执行对应的方法，需要将方法设置成为static
        void start() {
              //int n = pthread_create(&tid_,nullptr,start_rountine,ctx);
            int n = pthread_create(&tid_,nullptr,start_rountine,this);
            assert(n == 0);//编译debug的方式发布的时候存在，release方式发布，assert就不存在了，n就是一个定义了，但是没有被使用的变量
            // 在有些编译器下会有warning
            (void)n;
        }
        string threadname () {
            return name_;
        }
        void join() {
            int n = pthread_join(tid_,nullptr);           
            assert(n == 0);
            (void)n; 
        }
        void *callback() {
            return func_(args);
        }
        ~Thread()
        {
            //do nothing
        }
    private:
        string name_;
        func_t func_;
        void *args_;
        pthread_t tid_;
        static int threadnum = 1;
    };
    int Thread::threadnum = 1;
}
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
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85

(4)ThreadPool.hpp

(5)main.cc