• linux篇【11】：linux下的线程＜后序＞

    目录

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

    一.条件变量

    1.条件变量概念

    2.条件变量接口

    （1）pthread_cond_init  创建/释放 条件变量

    （2）pthread_cond_wait  阻塞线程

    （3）pthread_cond_signal 唤醒线程

    3.例子

    二.生产者消费者模型

    1.介绍

    2.生产者消费者模型优点

    3.基于BlockingQueue的生产者消费者模型

    阻塞队列概念

    4.BlockingQueue阻塞队列代码

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

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

    （3）易错

    （3）并发的体现

    makefile 新写法

    Task.hpp

    BlockQueue.hpp

    BlockQueueTest.cc

    三.POSIX信号量

    1.介绍

    2.信号量接口

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

    （2）sem_destroy 销毁一个信号量

     （3）sem_wait --信号量

    （4）sem_post ++信号量

    3.环形队列

    （1）信号量的作用

    4.信号量实现环形队列代码

    （1）单生产者，单消费者

    RingQueue.hpp

    RingQueueTest.cc

    （2）多生产者，多消费者

    Makefile

    RingQueue.hpp

    RingQueueTest.cc

    四.线程池

    1. STL,智能指针和线程安全

    2.线程池代码

    （1）线程池

    Log.hpp

    Task.hpp

    ThreadPool.hpp

    ThreadPool.cc

    （2）线程池单例模式

    Log.hpp

    Lock.hpp

    Task.hpp

    ThreadPool.hpp

    ThreadPoolTest.cc

    五.读者写者问题（了解）

    1.读写锁概念（rwlock）

    2.和生产消费者模型一样的321原则

    3.接口

    （1）初始化、销毁读写锁

     （2）加读者锁

    （3）加写者锁

    （4）释放锁

    4.特点

    5.测试代码

    makefile

    readerAndWriter.cc 

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

    不合理：互斥有可能导致饥饿问题——由于执行流1优先级高，他就不断的申请锁，释放锁，则另一个执行流2会长时间得不到某种资源

    在保证临界资源安全的前提下(互斥等)，让线程访问某种资源，具有一定的顺序性，称之为同步。

    同步：在保证数据安全的前提下，让线程能够按照某种特定的顺序访问临界资源，从而有效避免饥饿问题，叫做同步

    竞态条件：因为时序问题，而导致程序异常，我们称之为竞态条件。在线程场景下，这种问题也不难理解
    同步能够解决线程互斥不合理性的问题：防止饥饿，线程协同

    一.条件变量

    1.条件变量概念

    当一个线程互斥地访问某个变量时，它可能发现在其它线程改变状态之前，它什么也做不了。
    例如一个线程访问队列时，发现队列为空，它只能等待，直到其它线程将一个节点添加到队列中。这种情况就需要用到条件变量。

    条件(对应的共享资源的状态，程序员要判断资源是否满足自己操作的要求，为满/为空就不满足)，条件变量(条件满足或者不满足的时候，进行wait或signal--种方式)
    条件变量：通过判断条件是否满足要求来决定是否让当前线程等待。

    2.条件变量接口

    （1）pthread_cond_init  创建/释放 条件变量

    man 3 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  阻塞线程

    man 3 pthread_cond_wait

    pthread_cond_wait(&cond, &mutex);  让对应的线程进行等待，等待被唤醒。即调用这个接口线程会被阻塞。

     int pthread_cond_wait(pthread_cond_t *restrict cond,pthread_mutex_t *restrict mutex);
    条件变量要和mutex互斥锁，一并使用，为什么？——条件变量的wait中需要传入锁的意义是：在阻塞线程的时候，会自动释放mutex_锁（解锁）详解见：大标题二—>小标题4.(1)

    int pthread_cond_timedwait(pthread_cond_t *restrict cond,pthread_mutex_t *restrict mutex,
    const struct timespec *restrict abstime);        abstime设定时间（暂时不用这个接口）

    （3）pthread_cond_signal 唤醒线程

    man 3 pthread_cond_signal

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

    3.例子

    示例1：线程阻塞后被唤醒时是排队执行的。（这里是输入a就唤醒线程执行打印，q就退出）

    1. #include
    2. #include
    3. #include
    4. #include
    5. using namespace std;
    6.  
    7. pthread_cond_t cond;
    8. pthread_mutex_t mutex=PTHREAD_MUTEX_INITIALIZER;
    9. vectorvoid()>> funcs;
    10. void show()
    11. {
    12.     cout<<"hello show"<<" thread:"<<pthread_self()<
    13. }
    14. void print()
    15. {
    16.     cout<<"hello print"<
    17. }
    18. void* waitCommand(void* arg)
    19. {
    20.     pthread_detach(pthread_self());
    21.     //cout<<"!!!!!!!111111"<
    22.     while(true)
    23.     {
    24.         pthread_cond_wait(&cond,&mutex);
    25.         for(auto& f:funcs)
    26.         {
    27.             f();
    28.         }
    29.     }
    30.     cout<<"thread id: "<<pthread_self()<<" end... "<
    31.     return nullptr;
    32. }
    33. int main()
    34. {
    35.     funcs.push_back(show);
    36.     funcs.push_back(print);
    37.     funcs.push_back([](){
    38.         cout<<"你好，条件变量！"<
    39.     });
    40.     
    41.     pthread_cond_init(&cond,nullptr);
    42.  
    43.     pthread_t t1,t2,t3;
    44.     pthread_create(&t1,nullptr,waitCommand,nullptr);
    45.     pthread_create(&t2,nullptr,waitCommand,nullptr);
    46.     pthread_create(&t3,nullptr,waitCommand,nullptr);
    47.  
    48.     //cout<<"2222222222222222"<
    49.     while(true)
    50.     {
    51.         char n='a';
    52.         cout<<"请输入你的command(a/q)：";
    53.         cin>>n;
    54.         if(n=='a')
    55.         {
    56.             pthread_cond_signal(&cond);
    57.         }
    58.         else
    59.         {
    60.             break;
    61.         }
    62.     }
    63.     pthread_cond_destroy(&cond);
    64.     return 0;
    65. }

     

    示例2：线程阻塞后被唤醒时是排队执行的。（这里是不用输入一直循环唤醒线程执行打印）

    1. #include
    2. #include
    3. #include
    4. #include
    5. #include
    6. using namespace std;
    7.  
    8. pthread_cond_t cond;
    9. pthread_mutex_t mutex=PTHREAD_MUTEX_INITIALIZER;
    10. vectorvoid()>> funcs;
    11. void show()
    12. {
    13.     cout<<"hello show"<<" thread:"<<pthread_self()<
    14. }
    15. void print()
    16. {
    17.     cout<<"hello print"<
    18. }
    19. void* waitCommand(void* arg)
    20. {
    21.     pthread_detach(pthread_self());
    22.     //cout<<"!!!!!!!111111"<
    23.     while(true)
    24.     {
    25.         pthread_cond_wait(&cond,&mutex);
    26.         for(auto& f:funcs)
    27.         {
    28.             f();
    29.         }
    30.     }
    31.     cout<<"thread id: "<<pthread_self()<<" end... "<
    32.     return nullptr;
    33. }
    34. int main()
    35. {
    36.     funcs.push_back(show);
    37.     funcs.push_back(print);
    38.     funcs.push_back([](){
    39.         cout<<"你好，条件变量！"<
    40.     });
    41.     
    42.     pthread_cond_init(&cond,nullptr);
    43.  
    44.     pthread_t t1,t2,t3;
    45.     pthread_create(&t1,nullptr,waitCommand,nullptr);
    46.     pthread_create(&t2,nullptr,waitCommand,nullptr);
    47.     pthread_create(&t3,nullptr,waitCommand,nullptr);
    48.  
    49.     //cout<<"2222222222222222"<
    50.     while(true)
    51.     {
    52.         sleep(1);
    53.         pthread_cond_broadcast(&cond);
    54.     }
    55.     pthread_cond_destroy(&cond);
    56.     return 0;
    57. }

    二.生产者消费者模型

    1.介绍

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

    ①生产者和生产者(互斥)，消费者和消费者(互斥)，生产者和消费者( 互斥/同步)——3种关系
    ②生产者和消费者——由线程承担的 2种角色
    ③超市:内存中特定的一种内存结构(数据结构)——1个交易场所
            为了便于理解，我们就叫他 < 321原则 >
                    写代码，就是要完成321原则

    这里的超市就是缓冲区，也是临界资源：①提高效率。②解耦生产者和消费者之间的耦合关系。
    （一般就是内存中的一段空间，可以有自己的组织方式）

    1.消费者有多个，消费者之间是竞争关系——互斥关系
    2.生产者有多个，生产者之间也是竞争关系——互斥关系
    3.消费者和生产者之间又是什么关系呢?
    ①互斥关系：生产者把"123456789"写入缓冲区时（正在生产），消费者突然来拿，可能只拿走了"12345"就错误了，所以消费者和生产者之间也要有互斥关系。
    ②同步关系：要有一定的顺序去执行——消费完了要生产，生产满了要消费

    1.如何让多个消费者线程等待呢?又如何让消费者线程被唤醒呢?
    2.如何让多个生产者线程等待呢?又如何让生产者线程被唤醒呢?
    3.如何衡量消费者和生产者所关心的条件是否就绪呢??

    2.生产者消费者模型优点

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

    3.基于BlockingQueue的生产者消费者模型

    阻塞队列概念

    在多线程编程中阻塞队列 (Blocking Queue) 是一种常用于实现生产者和消费者模型的数据结构。其与普通的队列区别在于，当队列为空时，从队列获取元素的操作将会被阻塞，直到队列中被放入了元素；当队列满时，往队列里存放元素的操作也会被阻塞，直到有元素被从队列中取出( 以上的操作都是基于不同的线程来说的，线程在对阻塞队列进程操作时会被阻塞)，管道就是一种阻塞队列

    4.BlockingQueue阻塞队列代码

    注意：

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

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

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

    while (isFull())：原因是 proBlockWait();调用的 pthread_cond_wait(&proCond_, &mutex_); 有可能返回失败，或被 伪唤醒（伪唤醒可能是系统造成的或者写的代码有错误造成）。如果是 if (isFull()) 则返回失败/伪唤醒后 就直接向下指向条件满足的代码，可是此时阻塞队列还是满的，再添加数据就错了；所以用while (isFull()) 返回失败/伪唤醒后需要继续判断是否为满，为满就是返回失败/伪唤醒导致的—>重新等待；不满就是等待成功，消费者也消费了—>就可以添加数据了

    （3）易错

    ①pop是public。②BlockQueueTest.cc的consumer的get不知道怎么写/忘写。

    1. //生产接口
    2.     void push(const T &in) // const &: 纯输入
    3.     {
    4.         // 加锁
    5.         // 判断->是否适合生产->bq是否为满->程序员视角的条件->1. 满(不生产) 2. 不满(生产)
    6.         // if(满) 不生产,休眠
    7.         // else if(不满) 生产，唤醒消费者
    8.         // 解锁
    9.  
    10.         lockQueue();
    11.         while (isFull()) // ifFull就是我们在临界区中设定的条件
    12.         {
    13.             // before: 当我等待的时候，会自动释放mutex_
    14.             proBlockWait(); //阻塞等待，等待被唤醒。 被唤醒 ！= 条件被满足(概率虽然很小)，被唤醒 && 条件被满足
    15.             // after: 当我醒来的时候，我是在临界区里醒来的！！
    16.         }
    17.         // 条件满足，可以生产
    18.         pushCore(in); //生产完成
    19.         // wakeupCon(); // 唤醒消费者
    20.         unlockQueue();
    21.         wakeupCon(); // 唤醒消费者
    22.     }

    （3）并发的体现

    制作任务要花时间，处理任务也是要花时间的。并发体现：一般并不是在临界区中并发，而是生产前(生产者放入阻塞队列之前)，消费后(消费者从阻塞队列拿出之后)对应的并发。即：生产者放任务和消费者取任务是串行的，但是生产者制作任务和消费者处理任务是并发的。

    makefile 新写法

    makefile中定义变量，例如：LD=-lpthread ，=两侧尽量不带空格，下面就可以替换为 $(LD) ，意为取出这个变量

    1. CC=g++
    2. FLAGS=-std=c++11
    3. LD=-lpthread
    4. bin=blockQueue
    5. src=BlockQueueTest.cc
    6.  
    7. $(bin):$(src)
    8. 	$(CC) -o $@ $^ $(LD) $(FLAGS)
    9. .PHONY:clean
    10. clean:
    11. 	rm -f $(bin)

    Task.hpp

    1. #pragma once
    2.  
    3. #include 
    4. #include 
    5.  
    6. class Task
    7. {
    8. public:
    9.     Task() : elemOne_(0), elemTwo_(0), operator_('0')
    10.     {
    11.     }
    12.     Task(int one, int two, char op) : elemOne_(one), elemTwo_(two), operator_(op)
    13.     {
    14.     }
    15.     int operator() ()
    16.     {
    17.         return run();
    18.     }
    19.     int run()
    20.     {
    21.         int result = 0;
    22.         switch (operator_)
    23.         {
    24.         case '+':
    25.             result = elemOne_ + elemTwo_;
    26.             break;
    27.         case '-':
    28.             result = elemOne_ - elemTwo_;
    29.             break;
    30.         case '*':
    31.             result = elemOne_ * elemTwo_;
    32.             break;
    33.         case '/':
    34.         {
    35.             if (elemTwo_ == 0)
    36.             {
    37.                 std::cout << "div zero, abort" << std::endl;
    38.                 result = -1;
    39.             }
    40.             else
    41.             {
    42.                 result = elemOne_ / elemTwo_;
    43.             }
    44.         }
    45.  
    46.         break;
    47.         case '%':
    48.         {
    49.             if (elemTwo_ == 0)
    50.             {
    51.                 std::cout << "mod zero, abort" << std::endl;
    52.                 result = -1;
    53.             }
    54.             else
    55.             {
    56.                 result = elemOne_ % elemTwo_;
    57.             }
    58.         }
    59.         break;
    60.         default:
    61.             std::cout << "非法操作: " << operator_ << std::endl;
    62.             break;
    63.         }
    64.         return result;
    65.     }
    66.     int get(int *e1, int *e2, char *op)
    67.     {
    68.         *e1 = elemOne_;
    69.         *e2 = elemTwo_;
    70.         *op = operator_;
    71.     }
    72. private:
    73.     int elemOne_;
    74.     int elemTwo_;
    75.     char operator_;
    76. };

    BlockQueue.hpp

    1. #pragma once
    2.  
    3. #include 
    4. #include 
    5. #include 
    6. #include 
    7. #include 
    8.  
    9. using namespace std;
    10.  
    11. // 新需求: 我只想保存最新的5个任务，如果来了任务，老的任务，我想让他直接被丢弃
    12.  
    13. const uint32_t gDefaultCap = 5;
    14.  
    15. template <class T>
    16. class BlockQueue
    17. {
    18. public:
    19.     BlockQueue(uint32_t cap = gDefaultCap) : cap_(cap)
    20.     {
    21.         pthread_mutex_init(&mutex_, nullptr);
    22.         pthread_cond_init(&conCond_, nullptr);
    23.         pthread_cond_init(&proCond_, nullptr);
    24.     }
    25.     ~BlockQueue()
    26.     {
    27.         pthread_mutex_destroy(&mutex_);
    28.         pthread_cond_destroy(&conCond_);
    29.         pthread_cond_destroy(&proCond_);
    30.     }
    31.  
    32. public:
    33.     //生产接口
    34.     void push(const T &in) // const &: 纯输入
    35.     {
    36.         // 加锁
    37.         // 判断->是否适合生产->bq是否为满->程序员视角的条件->1. 满(不生产) 2. 不满(生产)
    38.         // if(满) 不生产,休眠
    39.         // else if(不满) 生产，唤醒消费者
    40.         // 解锁
    41.  
    42.         lockQueue();
    43.         while (isFull()) // ifFull就是我们在临界区中设定的条件
    44.         {
    45.             // before: 当我等待的时候，会自动释放mutex_
    46.             proBlockWait(); //阻塞等待，等待被唤醒。 被唤醒 ！= 条件被满足(概率虽然很小)，被唤醒 && 条件被满足
    47.             // after: 当我醒来的时候，我是在临界区里醒来的！！
    48.         }
    49.         // 条件满足，可以生产
    50.         pushCore(in); //生产完成
    51.         // wakeupCon(); // 唤醒消费者
    52.         unlockQueue();
    53.         wakeupCon(); // 唤醒消费者
    54.     }
    55.     //消费接口
    56.     T pop()
    57.     {
    58.         // 加锁
    59.         // 判断->是否适合消费->bq是否为空->程序员视角的条件->1. 空(不消费) 2. 有(消费)
    60.         // if(空) 不消费,休眠
    61.         // else if(有) 消费，唤醒生产者
    62.         // 解锁
    63.         lockQueue();
    64.         while (isEmpty())
    65.         {
    66.             conBlockwait(); //阻塞等待，等待被唤醒，？
    67.         }
    68.         // 条件满足，可以消费
    69.         T tmp = popCore();
    70.         unlockQueue();
    71.         wakeupPro(); // 唤醒生产者
    72.  
    73.         return tmp;
    74.     }
    75.  
    76. private:
    77.     void lockQueue()
    78.     {
    79.         pthread_mutex_lock(&mutex_);
    80.     }
    81.     void unlockQueue()
    82.     {
    83.         pthread_mutex_unlock(&mutex_);
    84.     }
    85.     bool isEmpty()
    86.     {
    87.         return bq_.empty();
    88.     }
    89.     bool isFull()
    90.     {
    91.         return bq_.size() == cap_;
    92.     }
    93.     void proBlockWait() // 生产者一定是在临界区中的！
    94.     {
    95.         // 1. 在阻塞线程的时候，会自动释放mutex_锁
    96.         pthread_cond_wait(&proCond_, &mutex_);
    97.     }
    98.  
    99.     void conBlockwait() //阻塞等待，等待被唤醒
    100.     {
    101.         // 1. 在阻塞线程的时候，会自动释放mutex_锁
    102.         pthread_cond_wait(&conCond_, &mutex_);
    103.         // 2. 当阻塞结束，返回的时候，pthread_cond_wait，会自动帮你重新获得mutex_，然后才返回
    104.         // 为什么我们上节课，写的代码，批量退出线程的时候，发现无法退出？
    105.     }
    106.  
    107.     void wakeupPro() // 唤醒生产者
    108.     {
    109.         pthread_cond_signal(&proCond_);
    110.     }
    111.     void wakeupCon() // 唤醒消费者
    112.     {
    113.         pthread_cond_signal(&conCond_);
    114.     }
    115.     void pushCore(const T &in)
    116.     {
    117.         bq_.push(in); //生产完成
    118.     }
    119.     T popCore()
    120.     {
    121.         T tmp = bq_.front();
    122.         bq_.pop();
    123.         return tmp;
    124.     }
    125.  
    126. private:
    127.     uint32_t cap_;           //容量
    128.     queue bq_;            // blockqueue
    129.     pthread_mutex_t mutex_;  //保护阻塞队列的互斥锁
    130.     pthread_cond_t conCond_; // 让消费者等待的条件变量
    131.     pthread_cond_t proCond_; // 让生产者等待的条件变量
    132. };

    BlockQueueTest.cc

    1. #include "Task.hpp"
    2. #include "BlockQueue.hpp"
    3.  
    4.  
    5. #include 
    6.  
    7. const std::string ops = "+-*/%";
    8.  
    9. // 并发，并不是在临界区中并发(一般)，而是生产前(before blockqueue)，消费后(after blockqueue)对应的并发
    10.  
    11. void *consumer(void *args)
    12. {
    13.     BlockQueue *bqp = static_cast *>(args);
    14.     while (true)
    15.     {
    16.         Task t = bqp->pop(); // 消费任务
    17.         int result = t();    //处理任务 --- 任务也是要花时间的！
    18.         int one, two;
    19.         char op;
    20.         t.get(&one, &two, &op);
    21.         cout << "consumer[" << pthread_self() << "] " << (unsigned long)time(nullptr) 
    22.         << " 消费了一个任务: " << one << op << two << "=" << result << endl;
    23.     }
    24. }
    25. void *productor(void *args)
    26. {
    27.     BlockQueue *bqp = static_cast *>(args);
    28.     while (true)
    29.     {
    30.         // 1. 制作任务 --- 要不要花时间？？ -- 网络，磁盘，用户
    31.         int one = rand() % 50;
    32.         int two = rand() % 20;
    33.         char op = ops[rand() % ops.size()];
    34.         Task t(one, two, op);
    35.         // 2. 生产任务
    36.         bqp->push(t);
    37.         cout << "producter[" << pthread_self() << "] " << (unsigned long)time(nullptr) 
    38.         << " 生产了一个任务: " << one << op << two << "=?" << endl;
    39.         sleep(1);
    40.     }
    41. }
    42.  
    43. int main()
    44. {
    45.     srand((unsigned long)time(nullptr) ^ getpid());
    46.     // 定义一个阻塞队列
    47.     // 创建两个线程，productor, consumer
    48.     // productor -----  consumer
    49.     // BlockQueue bq;
    50.     // bq.push(10);
    51.     // int a = bq.pop();
    52.     // cout << a << endl;
    53.     // 既然可以使用int类型的数据，我们也可以使用自己封装的类型，包括任务
    54.     // BlockQueue bq;
    55.     BlockQueue bq;
    56.  
    57.     pthread_t c, p;
    58.     pthread_create(&c, nullptr, consumer, &bq);
    59.     pthread_create(&p, nullptr, productor, &bq);
    60.  
    61.     pthread_join(c, nullptr);
    62.     pthread_join(p, nullptr);
    63.     return 0;
    64. }

     

    三.POSIX信号量

    1.介绍

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

    信号量本质是一个计数器。定义：信号量是一个计数器，这个计数器对应的PV操作(V ++ P --)是原子的。信号量的PV操作：V ++归还资源，P --申请资源。信号量的作用：限制进入临界区的线程个数

    申请mutex 访问临界资源之前被切换走了有问题吗?——没问题。资源预定:：因为只要我拿到了锁，当前的临界资源就是我的!
    1.信号量申请成功了，就一定能保证你会拥有一部分临界资源吗?——能，只要信号量申请成功，那么你一定会获得指定的资源
    2.临界资源可以当成整体，可以不可以看做一小部分一小部分呢?——可以，结合场景

    信号量: 1
    p--   1—>0---加锁
    v++ 0—>1---释放锁
    叫做二元信号量，二元信号量 和 互斥锁 是等价的

    2.信号量接口

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

     man 3 sem_init

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

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

    value ：信号量初始值

    （2）sem_destroy 销毁一个信号量

    int sem_destroy(sem_t *sem);

     （3）sem_wait --信号量

    功能：等待信号量，会将信号量的值减1
    int sem_wait(sem_t *sem); ——P--

    （4）sem_post ++信号量

    功能：发布信号量，表示资源使用完毕，可以归还资源了。将信号量值加1 。
    int sem_post(sem_t *sem); ——V++
    上面生产者 - 消费者的例子是基于 queue 的 , 其空间可以动态分配 , 现在基于固定大小的环形队列重写这个程序 （POSIX 信号量）

    3.环形队列

    （1）信号量的作用

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

    环形队列有可能访问同一个位置。什么时候会发生?——我们两个指向同一个位置的时候只有满or空的时候! ( 互斥and同步)其他时候，都指向不同的位置! (并发 )
    1.数据为空：消费者不能超过生产者一>生产者先运行。
    2.数据为满：生产者不能把消费者套一个圈然后继续再往后写入——消费者先运行 解释如下：
    生产者:最关心的是什么资源：空间默认是N: [N, 0]
    消费者:最关心的是什么资源：数据默认是0 [0，N]

    sem_ t roomSem=N; 剩余空间的信号量
    sem_ t dataSem=0; 数据的信号量

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

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

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

    我们是单生产者，单消费者

    多生产者，多消费者？？代码怎么改？

    为什么呢？？？多生产者，多消费者？

    不要只关心把数据或者任务，从ringqueue 放拿的过程，获取数据或者任务，处理数据或者任务，也是需要花时间的！

    4.信号量实现环形队列代码

    （1）单生产者，单消费者

    RingQueue.hpp

    1. #include
    2. #include
    3. #include
    4. #include
    5. #include
    6. using namespace std;
    7.  
    8. const int gCap=5;
    9. template<class T>
    10. class RingQueue
    11. {
    12. public:
    13.     RingQueue(int cap=gCap):ringqueue_(cap),pIndex_(0),cIndex_(0)
    14.     {
    15.         sem_init(&roomSem_,0,ringqueue_.size());
    16.         sem_init(&dataSem_,0,0);
    17.     }
    18.     void push(const T& in)
    19.     {
    20.         sem_wait(&roomSem_);
    21.         ringqueue_[pIndex_]=in;
    22.         sem_post(&dataSem_);
    23.         pIndex_++;
    24.         pIndex_%=ringqueue_.size();
    25.     }
    26.     T pop()
    27.     {
    28.         sem_wait(&dataSem_);
    29.         T temp=ringqueue_[cIndex_];
    30.         sem_post(&roomSem_);
    31.         cIndex_++;
    32.         cIndex_%=ringqueue_.size();
    33.         return temp;
    34.     }
    35.     ~RingQueue()
    36.     {
    37.         sem_destroy(&roomSem_);
    38.         sem_destroy(&dataSem_);
    39.     }
    40. private:
    41.     vector ringqueue_;
    42.     sem_t roomSem_; 
    43.     sem_t dataSem_;
    44.     uint32_t cIndex_;
    45.     uint32_t pIndex_;
    46. };

    RingQueueTest.cc

    1. #include "RingQueue.hpp"
    2.  
    3. void *producer(void *args)
    4. {
    5.     RingQueue<int> *rqp = static_castint> *>(args);
    6.     while (true)
    7.     {
    8.         int data = rand() % 10;
    9.         rqp->push(data);
    10.         cout << "Propthread[" << pthread_self() << "]"
    11.              << "生产了一个数据：" << data << endl;
    12.              sleep(1);
    13.     }
    14. }
    15.  
    16. void *consumer(void *args)
    17. {
    18.     RingQueue<int> *rqp = static_castint> *>(args);
    19.     while (true)
    20.     {
    21.         int data = rqp->pop();
    22.         cout << "Conpthread[" << pthread_self() << "]"
    23.              << "消费了一个数据：" << data << endl;
    24.     }
    25. }
    26. int main()
    27. {
    28.     srand((unsigned int)time(nullptr) ^ getpid());
    29.     RingQueue<int> rq;
    30.     pthread_t c1, p1;
    31.     pthread_create(&p1, nullptr, producer, &rq);
    32.     pthread_create(&c1, nullptr, consumer, &rq);
    33.     pthread_join(c1, nullptr);
    34.     pthread_join(p1, nullptr);
    35.     return 0;
    36. }

    （2）多生产者，多消费者

    与单线程的区别：要上锁，使生产者与生产者之间生产互斥；消费者与消费者之间消费互斥。如果 sem_wait(&roomSem_);P--操作放在lock后面，sem_post(&dataSem_); 放在unlock前面，每次只能进入一个线程申请信号量，则信号量无法被多次的申请，只能互斥申请。但信号量本身就是资源的预定机制，就应该让多个资源一起申请，然后再一个一个排队执行锁内的生产过程。申请信号量是原子的，生产过程是非原子的，所以生产过程要被保护，消费过程同理。

    1.     // 生产
    2.     void push(const T &in)
    3.     {
    4.         sem_wait(&roomSem_); 
    5.         pthread_mutex_lock(&pmutex_);
    6.         //sem_wait(&roomSem_); //如果放在lock后面，则信号量无法被多次的申请
    7.         ringqueue_[pIndex_] = in; //生产的过程
    8.         pIndex_++;   // 写入位置后移
    9.         pIndex_ %= ringqueue_.size(); // 更新下标，保证环形特征
    10.         //sem_post(&dataSem_); //如果放在unlock前面
    11.         pthread_mutex_unlock(&pmutex_);
    12.         sem_post(&dataSem_);
    13.     }

    Makefile

    1. CC=g++
    2. FLAGS=-std=c++11
    3. LD=-lpthread
    4. bin=ringQueue
    5. src=RingQueueTest.cc
    6.  
    7. $(bin):$(src)
    8. 	$(CC) -o $@ $^ $(LD) $(FLAGS)
    9. .PHONY:clean
    10. clean:
    11. 	rm -f $(bin)

    RingQueue.hpp

    1. #pragma once
    2.  
    3. #include 
    4. #include 
    5. #include 
    6. #include 
    7.  
    8. using namespace std;
    9.  
    10. const int gCap = 10;
    11.  
    12. template <class T>
    13. class RingQueue
    14. {
    15. public:
    16.     RingQueue(int cap = gCap): ringqueue_(cap), pIndex_(0), cIndex_(0)
    17.     {
    18.         // 生产
    19.         sem_init(&roomSem_, 0, ringqueue_.size());
    20.         // 消费
    21.         sem_init(&dataSem_, 0, 0);
    22.  
    23.         pthread_mutex_init(&pmutex_ ,nullptr);
    24.         pthread_mutex_init(&cmutex_ ,nullptr);
    25.     }
    26.     // 生产
    27.     void push(const T &in)
    28.     {
    29.         sem_wait(&roomSem_); //无法被多次的申请
    30.         pthread_mutex_lock(&pmutex_);
    31.  
    32.         ringqueue_[pIndex_] = in; //生产的过程
    33.         pIndex_++;   // 写入位置后移
    34.         pIndex_ %= ringqueue_.size(); // 更新下标，保证环形特征
    35.  
    36.         pthread_mutex_unlock(&pmutex_);
    37.         sem_post(&dataSem_);
    38.     }
    39.     // 消费
    40.     T pop()
    41.     {
    42.         sem_wait(&dataSem_);
    43.         pthread_mutex_lock(&cmutex_);
    44.  
    45.         T temp = ringqueue_[cIndex_];
    46.         cIndex_++;
    47.         cIndex_ %= ringqueue_.size();// 更新下标，保证环形特征
    48.  
    49.         pthread_mutex_unlock(&cmutex_);
    50.         sem_post(&roomSem_);
    51.  
    52.         return temp;
    53.     }
    54.     ~RingQueue()
    55.     {
    56.         sem_destroy(&roomSem_);
    57.         sem_destroy(&dataSem_);
    58.  
    59.         pthread_mutex_destroy(&pmutex_);
    60.         pthread_mutex_destroy(&cmutex_);
    61.     }
    62. private:
    63.     vector ringqueue_; // 唤醒队列
    64.     sem_t roomSem_;       // 衡量空间计数器，productor
    65.     sem_t dataSem_;       // 衡量数据计数器，consumer
    66.     uint32_t pIndex_;     // 当前生产者写入的位置， 如果是多线程，pIndex_也是临界资源
    67.     uint32_t cIndex_;     // 当前消费者读取的位置，如果是多线程，cIndex_也是临界资源
    68.  
    69.     pthread_mutex_t pmutex_;
    70.     pthread_mutex_t cmutex_;
    71. };

    RingQueueTest.cc

    1. #include "RingQueue.hpp"
    2. #include 
    3. #include 
    4.  
    5. // 我们是单生产者，单消费者
    6. // 多生产者，多消费者？？代码怎么改？
    7. // 为什么呢？？？多生产者，多消费者？
    8. // 不要只关心把数据或者任务，从ringqueue 放拿的过程，获取数据或者任务，处理数据或者任务，也是需要花时间的！
    9.  
    10. void *productor(void *args)
    11. {
    12.     RingQueue<int> *rqp = static_castint> *>(args);
    13.     while(true)
    14.     {
    15.         int data = rand()%10;
    16.         rqp->push(data);
    17.         cout << "pthread[" << pthread_self() << "]" << " 生产了一个数据: " << data << endl;
    18.         //sleep(1);
    19.     }
    20. }
    21.  
    22. void *consumer(void *args)
    23. {
    24.     RingQueue<int> *rqp = static_castint> *>(args);
    25.     while(true)
    26.     {
    27.         sleep(10);
    28.         int data = rqp->pop();
    29.         cout << "pthread[" << pthread_self() << "]" << " 消费了一个数据: " << data << endl;
    30.     }
    31. }
    32.  
    33. int main()
    34. {
    35.     srand((unsigned long)time(nullptr)^getpid());
    36.  
    37.     RingQueue<int> rq;
    38.  
    39.     pthread_t c1,c2,c3, p1,p2,p3;
    40.     pthread_create(&p1, nullptr, productor, &rq);
    41.     pthread_create(&p2, nullptr, productor, &rq);
    42.     pthread_create(&p3, nullptr, productor, &rq);
    43.     pthread_create(&c1, nullptr, consumer, &rq);
    44.     pthread_create(&c2, nullptr, consumer, &rq);
    45.     pthread_create(&c3, nullptr, consumer, &rq);
    46.  
    47.  
    48.     pthread_join(c1, nullptr);
    49.     pthread_join(c2, nullptr);
    50.     pthread_join(c3, nullptr);
    51.     pthread_join(p1, nullptr);
    52.     pthread_join(p2, nullptr);
    53.     pthread_join(p3, nullptr);
    54.  
    55.     return 0;
    56. }

    四.线程池

    1. STL,智能指针和线程安全

    STL 中的容器是否是线程安全的 ?
    不是 .
    原因是 , STL 的设计初衷是将性能挖掘到极致 , 而一旦涉及到加锁保证线程安全 , 会对性能造成巨大的影响 .
    而且对于不同的容器 , 加锁方式的不同 , 性能可能也不同 ( 例如 hash 表的锁表和锁桶 ).
    因此 STL 默认不是线程安全 . 如果需要在多线程环境下使用 , 往往需要调用者自行保证线程安全 .
    智能指针是否是线程安全的 ?
    对于 unique_ptr, 由于只是在当前代码块范围内生效 , 因此不涉及线程安全问题 .
    对于 shared_ptr, 多个对象需要共用一个引用计数变量 , 所以会存在线程安全问题 . 但是标准库实现的时候考虑到了这个问题, 基于原子操作 (CAS) 的方式保证 shared_ptr 能够高效 , 原子的操作引用计数.（ shared_ptr是线程安全的 ）
    12. 其他常见的各种锁
    悲观锁：在每次取数据时，总是担心数据会被其他线程修改，所以会在取数据前先加锁（读锁，写锁，行锁等），当其他线程想要访问数据时，被阻塞挂起。
    乐观锁：每次取数据时候，总是乐观的认为数据不会被其他线程修改，因此不上锁。但是在更新数据前，会判断其他数据在更新前有没有对数据进行修改。主要采用两种方式：版本号机制和CAS 操作。
    CAS 操作：当需要更新数据时，判断当前内存值和之前取得的值是否相等。如果相等则用新值更新。若不等则失败，失败则重试，一般是一个自旋的过程，即不断重试。
    自旋锁（ phtread_spin_init/lock ），公平锁，非公平锁

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

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

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

    2.线程池

    （1）概念

    线程池：对执行流的预先分配，当有任务时会直接指定线程去做，而不需要再创建线程。

    （2）特点

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

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

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

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

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

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

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

    /*threadpool.h*/
    /* 线程池 :
    *         一种线程使用模式。线程过多会带来调度开销，进而影响缓存局部性和整体性能。而线程池维护着多个线程，等待着
    监督管理者分配可并发执行的任务。这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能够保证内核的充分利
    用，还能防止过分调度。可用线程数量应该取决于可用的并发处理器、处理器内核、内存、网络 sockets 等的数量。
    * 线程池的应用场景：
    *         1. 需要大量的线程来完成任务，且完成任务的时间比较短。 WEB 服务器完成网页请求这样的任务，使用线程池技
    术是非常合适的。因为单个任务小，而任务数量巨大，你可以想象一个热门网站的点击次数。 但对于长时间的任务，比如一个
    Telnet 连接请求，线程池的优点就不明显了。因为 Telnet 会话时间比线程的创建时间大多了。
    *         2. 对性能要求苛刻的应用，比如要求服务器迅速响应客户请求。
    *         3. 接受突发性的大量请求，但不至于使服务器因此产生大量线程的应用。突发性大量客户请求，在没有线程池情
    况下，将产生大量线程，虽然理论上大部分操作系统线程数目最大值不是问题，短时间内产生大量线程可能使内存到达极限，
    出现错误 .
    * 线程池的种类：
    * 线程池示例：
    *         1. 创建固定数量线程池，循环从任务队列中获取任务对象，
    *         2. 获取到任务对象后，执行任务对象中的任务接口
    */

    3.线程池代码

    （1）线程池

    易错点：① start()中assert忘写

    ②线程池用智能指针维护

    ③因为类内的函数threadRoutine多传了一个this指针，所以和pthread_ create(&temp, nullptr, threadRoutine, this);里面的threadRoutine类型不匹配了，所以threadRoutine函数定义成static去掉隐含this才能类型匹配，并且pthread_ create传入this指针

    ④choiceThreadForHandler(); 选择一个线程去执行

    1.     void push(const T &in)
    2.     {
    3.         lockQueue();
    4.         taskQueue_.push(in);
    5.         choiceThreadForHandler();
    6.         unlockQueue();
    7.     }

    Log.hpp

    1. #pragma once
    2.  
    3. #include 
    4. #include 
    5. #include 
    6.  
    7. std::ostream &Log()
    8. {
    9.     std::cout << "Fot Debug |" << " timestamp: " << (uint64_t)time(nullptr) << " | " << " Thread[" << pthread_self() << "] | ";
    10.     return std::cout;
    11. }

    Task.hpp

    1. #pragma once
    2. #include
    3.  
    4. class Task
    5. {
    6. public:
    7.     Task() : elemOne_(0), elemTwo_(0), operator_('0')
    8.     {
    9.     }
    10.     Task(int one, int two, char op) : elemOne_(one), elemTwo_(two), operator_(op)
    11.     {
    12.     }
    13.     int operator()()
    14.     {
    15.         return run();
    16.     }
    17.     int run()
    18.     {
    19.         int result = 0;
    20.         switch (operator_)
    21.         {
    22.         case '+':
    23.             result = elemOne_ + elemTwo_;
    24.             break;
    25.         case '-':
    26.             result = elemOne_ - elemTwo_;
    27.             break;
    28.         case '*':
    29.             result = elemOne_ * elemTwo_;
    30.             break;
    31.         case '/':
    32.         {
    33.             if(elemTwo_==0)
    34.             {
    35.                 std::cout<<"div zero,abort"<
    36.                 result= -1;
    37.             }
    38.             else
    39.             {
    40.                 result = elemOne_ / elemTwo_;
    41.             }
    42.         }
    43.             break;
    44.         case '%':
    45.         {
    46.             if(elemTwo_==0)
    47.             {
    48.                 std::cout<<"mod zero,abort"<
    49.                 result= -1;
    50.             }
    51.             else
    52.             {
    53.                 result = elemOne_ % elemTwo_;
    54.             }
    55.         }
    56.             break;
    57.         default:
    58.             std::cout<<"非法操作: " <
    59.             break;
    60.         }
    61.         return result;
    62.     }
    63.     void get(int* e1,int* e2,char* op)
    64.     {
    65.         *e1=elemOne_;
    66.         *e2=elemTwo_;
    67.         *op=operator_;
    68.     }
    69. private:
    70.     int elemOne_;
    71.     int elemTwo_;
    72.     char operator_;
    73. };

    ThreadPool.hpp

    1. #include"Log.hpp"
    2. #include
    3. #include
    4. #include
    5. #include
    6. #include
    7. #include
    8. #include
    9. using namespace std;
    10. #include
    11.  
    12. int gThreadNum=5;
    13. template<class T>
    14. class ThreadPool
    15. {
    16. public:
    17.     ThreadPool(int threadNum=gThreadNum):threadNum_(threadNum),isStart_(false)
    18.     {
    19.         pthread_mutex_init(&mutex_,nullptr);
    20.         pthread_cond_init(&cond_,nullptr);
    21.     }
    22.     ~ThreadPool()
    23.     {
    24.         pthread_mutex_destroy(&mutex_);
    25.         pthread_cond_destroy(&cond_);
    26.     }
    27.  
    28.     void push(const T& in)
    29.     {
    30.         lockQueue();
    31.         taskQueue_.push(in);
    32.         choiceThreadForHander();
    33.         unlockQueue();
    34.         //sleep(1);
    35.     }
    36.     static void* threadRoutine(void* args)
    37.     {
    38.         pthread_detach(pthread_self());
    39.         ThreadPool* tp=static_cast*>(args);
    40.         while(true)
    41.         {
    42.             tp->lockQueue();
    43.             while(!tp->haveTask())    //这里用while：防止条件变量未唤醒的情况
    44.             {
    45.                 tp->waitForTask();
    46.             }
    47.             T t=tp->pop();
    48.             tp->unlockQueue();
    49.             int one,two;
    50.             char oper;
    51.             t.get(&one,&two,&oper);
    52.             Log()<<"新线程完成计算任务:"<"="<run()<
    53.             //cout<<"pthread["<
    54.         }
    55.     }
    56.     void start()
    57.     {
    58.         assert(!isStart_);      //防止线程池被重复启动
    59.         for(int i=0;i
    60.         {
    61.             pthread_t temp;
    62.             pthread_create(&temp,nullptr,threadRoutine,(void*)this);
    63.         }
    64.         isStart_=true;
    65.     }
    66. private:
    67.     T pop()
    68.     {
    69.         T temp=taskQueue_.front();
    70.         taskQueue_.pop();
    71.         return temp;
    72.     }
    73.     bool haveTask()
    74.     {
    75.         return !taskQueue_.empty();
    76.     }
    77.     void waitForTask()
    78.     {
    79.         pthread_cond_wait(&cond_,&mutex_);
    80.     }
    81.     void lockQueue()
    82.     {
    83.         pthread_mutex_lock(&mutex_);
    84.     }
    85.     void unlockQueue()
    86.     {
    87.         pthread_mutex_unlock(&mutex_);
    88.     }
    89.     void choiceThreadForHander()
    90.     {
    91.         pthread_cond_signal(&cond_);
    92.     }
    93.  
    94. private:
    95.     bool isStart_;
    96.     int threadNum_;
    97.     queue taskQueue_;
    98.     pthread_mutex_t mutex_;
    99.     pthread_cond_t cond_;
    100. };

    ThreadPool.cc

    1. #include"ThreadPool.hpp"
    2. #include"Task.hpp"
    3.  
    4.  
    5. int main()
    6. {
    7.     const string operators ="+/*/%";
    8.     unique_ptr> tp(new ThreadPool());
    9.     tp->start();
    10.     srand((unsigned int)time(nullptr)^getpid()^pthread_self());
    11.     while(true)
    12.     {
    13.         int one=rand()%50;
    14.         int two=rand()%50;
    15.         char oper=operators[rand()%operators.size()];
    16.         Log()<<"主线程派发计算任务："<"=?"<
    17.         Task t(one,two,oper);
    18.         tp->push(t);
    19.         //sleep(1);
    20.     }
    21.     
    22.     //while(1);
    23.     return 0;
    24. }

    （2）线程池单例模式

    变化：①私有化构造（析构不用私有，设为公有即可，仅仅把构造函数私有化就可以实现单例模式，单例对象也需要） ②delete了 operator=，拷贝构造。③多了静态成员变量instance。

    Log.hpp

    1. #pragma once
    2.  
    3. #include 
    4. #include 
    5. #include 
    6.  
    7. std::ostream &Log()
    8. {
    9.     std::cout << "Fot Debug |" << " timestamp: " << (uint64_t)time(nullptr) << " | " << " Thread[" << pthread_self() << "] | ";
    10.     return std::cout;
    11. }

    Lock.hpp

    1. #pragma once
    2.  
    3. #include 
    4. #include 
    5.  
    6. class Mutex
    7. {
    8. public:
    9.     Mutex()
    10.     {
    11.         pthread_mutex_init(&lock_, nullptr);
    12.     }
    13.     void lock()
    14.     {
    15.         pthread_mutex_lock(&lock_);
    16.     }
    17.     void unlock()
    18.     {
    19.         pthread_mutex_unlock(&lock_);
    20.     }
    21.     ~Mutex()
    22.     {
    23.         pthread_mutex_destroy(&lock_);
    24.     }
    25.  
    26. private:
    27.     pthread_mutex_t lock_;
    28. };
    29.  
    30. class LockGuard
    31. {
    32. public:
    33.     LockGuard(Mutex *mutex) : mutex_(mutex)
    34.     {
    35.         mutex_->lock();
    36.         std::cout << "加锁成功..." << std::endl;
    37.     }
    38.  
    39.     ~LockGuard()
    40.     {
    41.         mutex_->unlock();
    42.         std::cout << "解锁成功...." << std::endl;
    43.     }
    44.  
    45. private:
    46.     Mutex *mutex_;
    47. };

    Task.hpp

    1. #pragma once
    2.  
    3. #include 
    4. #include 
    5.  
    6. class Task
    7. {
    8. public:
    9.     Task() : elemOne_(0), elemTwo_(0), operator_('0')
    10.     {
    11.     }
    12.     Task(int one, int two, char op) : elemOne_(one), elemTwo_(two), operator_(op)
    13.     {
    14.     }
    15.     int operator() ()
    16.     {
    17.         return run();
    18.     }
    19.     int run()
    20.     {
    21.         int result = 0;
    22.         switch (operator_)
    23.         {
    24.         case '+':
    25.             result = elemOne_ + elemTwo_;
    26.             break;
    27.         case '-':
    28.             result = elemOne_ - elemTwo_;
    29.             break;
    30.         case '*':
    31.             result = elemOne_ * elemTwo_;
    32.             break;
    33.         case '/':
    34.         {
    35.             if (elemTwo_ == 0)
    36.             {
    37.                 std::cout << "div zero, abort" << std::endl;
    38.                 result = -1;
    39.             }
    40.             else
    41.             {
    42.                 result = elemOne_ / elemTwo_;
    43.             }
    44.         }
    45.  
    46.         break;
    47.         case '%':
    48.         {
    49.             if (elemTwo_ == 0)
    50.             {
    51.                 std::cout << "mod zero, abort" << std::endl;
    52.                 result = -1;
    53.             }
    54.             else
    55.             {
    56.                 result = elemOne_ % elemTwo_;
    57.             }
    58.         }
    59.         break;
    60.         default:
    61.             std::cout << "非法操作: " << operator_ << std::endl;
    62.             break;
    63.         }
    64.         return result;
    65.     }
    66.     int get(int *e1, int *e2, char *op)
    67.     {
    68.         *e1 = elemOne_;
    69.         *e2 = elemTwo_;
    70.         *op = operator_;
    71.     }
    72. private:
    73.     int elemOne_;
    74.     int elemTwo_;
    75.     char operator_;
    76. };

    ThreadPool.hpp

    1. #pragma once
    2.  
    3. #include 
    4. #include 
    5. #include 
    6. #include 
    7. #include 
    8. #include 
    9. #include 
    10. #include 
    11. #include "Log.hpp"
    12. #include "Lock.hpp"
    13.  
    14. using namespace std;
    15.  
    16. int gThreadNum = 5;
    17.  
    18. template <class T>
    19. class ThreadPool
    20. {
    21. private:
    22.     ThreadPool(int threadNum = gThreadNum) : threadNum_(threadNum), isStart_(false)
    23.     {
    24.         assert(threadNum_ > 0);
    25.         pthread_mutex_init(&mutex_, nullptr);
    26.         pthread_cond_init(&cond_, nullptr);
    27.     }
    28.     ThreadPool(const ThreadPool &) = delete;
    29.     void operator=(const ThreadPool&) = delete;
    30.  
    31. public:
    32.     static ThreadPool *getInstance()
    33.     {
    34.         static Mutex mutex;
    35.         if (nullptr == instance) //仅仅是过滤重复的判断
    36.         {
    37.             LockGuard lockguard(&mutex); //进入代码块，加锁。退出代码块，自动解锁
    38.             if (nullptr == instance)
    39.             {
    40.                 instance = new ThreadPool();
    41.             }
    42.         }
    43.  
    44.         return instance;
    45.     }
    46.     //类内成员, 成员函数，都有默认参数this
    47.     static void *threadRoutine(void *args)
    48.     {
    49.         pthread_detach(pthread_self());
    50.         ThreadPool *tp = static_cast *>(args);
    51.         prctl(PR_SET_NAME, "follower");
    52.         while (1)
    53.         {
    54.             tp->lockQueue();
    55.             while (!tp->haveTask())
    56.             {
    57.                 tp->waitForTask();    //无数据就在条件变量下排队等待
    58.             }
    59.             //这个任务就被拿到了线程的上下文中
    60.             T t = tp->pop();
    61.             tp->unlockQueue();
    62.  
    63.             // for debug
    64.             int one, two;
    65.             char oper;
    66.             t.get(&one, &two, &oper);
    67.             //规定，所有的任务都必须有一个run方法
    68.             Log() << "新线程完成计算任务: " << one << oper << two << "=" << t.run() << "\n";
    69.         }
    70.     }
    71.     void start()
    72.     {
    73.         assert(!isStart_);    //防止线程池被重复启动
    74.         for (int i = 0; i < threadNum_; i++)
    75.         {
    76.             pthread_t temp;
    77.             pthread_create(&temp, nullptr, threadRoutine, this);
    78.         }
    79.         isStart_ = true;
    80.     }
    81.     void push(const T &in)
    82.     {
    83.         lockQueue();
    84.         taskQueue_.push(in);
    85.         choiceThreadForHandler();//指定某个线程去执行，即：唤醒某个线程
    86.         unlockQueue();
    87.     }
    88.     ~ThreadPool()
    89.     {
    90.         pthread_mutex_destroy(&mutex_);
    91.         pthread_cond_destroy(&cond_);
    92.     }
    93.  
    94. private:
    95.     void lockQueue() { pthread_mutex_lock(&mutex_); }
    96.     void unlockQueue() { pthread_mutex_unlock(&mutex_); }
    97.     bool haveTask() { return !taskQueue_.empty(); }    //有线程就是true
    98.     void waitForTask() { pthread_cond_wait(&cond_, &mutex_); }
    99.     void choiceThreadForHandler() { pthread_cond_signal(&cond_); }
    100.     T pop()
    101.     {
    102.         T temp = taskQueue_.front();
    103.         taskQueue_.pop();
    104.         return temp;
    105.     }
    106.  
    107. private:
    108.     bool isStart_;    //表示当前线程池是否已经启动
    109.     int threadNum_;   //线程个数
    110.     queue taskQueue_;
    111.     pthread_mutex_t mutex_;
    112.     pthread_cond_t cond_;
    113.  
    114.     static ThreadPool *instance;
    115.     // const static int a = 100;
    116. };
    117.  
    118. template <class T>
    119. ThreadPool *ThreadPool::instance = nullptr;

    ThreadPoolTest.cc

    1. #include "ThreadPool.hpp"
    2. #include "Task.hpp"
    3. #include 
    4. #include 
    5.  
    6.  
    7. // 如何对一个线程进行封装, 线程需要一个回调函数，支持lambda
    8. // class tread{
    9. // };
    10.  
    11. int main()
    12. {
    13.     prctl(PR_SET_NAME, "master");    //给线程加名字
    14.  
    15.     const string operators = "+/*/%";
    16.     // unique_ptr > tp(new ThreadPool());
    17.     unique_ptr > tp(ThreadPool::getInstance());
    18.     tp->start();
    19.  
    20.     srand((unsigned long)time(nullptr) ^ getpid() ^ pthread_self());
    21.     // 派发任务的线程
    22.     while(true)
    23.     {
    24.         int one = rand()%50;
    25.         int two = rand()%10;
    26.         char oper = operators[rand()%operators.size()];
    27.         Log() << "主线程派发计算任务: " << one << oper << two << "=?" << "\n";
    28.         Task t(one, two, oper);
    29.         tp->push(t);
    30.         sleep(1);
    31.     }
    32. }

     

    五.读者写者问题（了解）

    1.读写锁概念（rwlock）

    在编写多线程的时候，有一种情况是十分常见的。那就是，有些公共数据修改的机会比较少。相比较改写，它们读的机会反而高的多。通常而言，在读的过程中，往往伴随着查找的操作，中间耗时很长。给这种代码段加锁，会极大地降低我们程序的效率。那么有没有一种方法，可以专门处理这种多读少写的情况呢？ 有，那就是读写锁。

    2.和生产消费者模型一样的321原则

    321原则
    2：读者和写者
    1：1个读写场所
    3：写者和写者——互斥；读者和读者——没有关系，可以并发执行；读者和写者——互斥关系
    读者写者vs生产者消费者本质区别：消费者会把数据拿走，而读者不会!

    读者：写者=n：1
            读者                        写者
            加读锁                    加写锁
            读取内容                写入修改内容
            释放锁                     释放锁

    3.接口

    （把mutex改成rwlock即可）

    （1）初始化、销毁读写锁

    man 3 pthread_rwlock_init

     （2）加读者锁

    读者写者用的同一把锁，只是身份不同，执行的代码不同

    man 3 pthread_rwlock_rdlock

    int pthread_rwlock_rdlock(pthread_rwlock_t *rwlock); 阻塞加锁

    int pthread_rwlock_tryrdlock(pthread_rwlock_t *rwlock); 非阻塞加锁

    读者 pthread_ rwlock_ rdlock () 原理：
    lock (mutex)
    readers++;
    unlock (mutex)
     read操作
    lock (mutex)
    readers-- ;
    unlock (mutex)

    （3）加写者锁

    读者写者用的同一把锁，只是身份不同，执行的代码不同

    man 3 pthread_rwlock_wrlock

    写者pthread_ rwlock_ wrlock()原理：
    lock (mutex)
    while (readers>0)释放锁，wait（wait会自动释放锁）
    write操作
    unlock (mutex) ;

    （4）释放锁

    man 3 pthread_rwlock_unlock

    4.特点

    读者写者进行操作的时候，读者非常多，频率个别高；写者比较少，频率不高。
    两种简单的策略：读者优先（默认策略） 或 写者优先（让当前读者读完，之后的读者阻塞，让写者先进入写）

    5.测试代码

    makefile

    1. readerAndWriter:readerAndWriter.cc
    2. 	g++ -o $@ $^ -lpthread -std=c++11
    3. .PHONY:clean
    4. clean:
    5. 	rm -f readerAndWriter

    readerAndWriter.cc 

    1. #include 
    2. #include 
    3. #include 
    4.  
    5. int board = 0;
    6.  
    7. pthread_rwlock_t rw;
    8.  
    9. using namespace std;
    10.  
    11. void *reader(void* args)
    12. {
    13.     const char *name = static_cast<const char *>(args);
    14.     cout << "run..." << endl;
    15.     while(true)
    16.     {
    17.         pthread_rwlock_rdlock(&rw);
    18.         cout << "reader read : " << board << "tid: " << pthread_self() << endl;
    19.         sleep(10);
    20.         pthread_rwlock_unlock(&rw);
    21.     }
    22. }
    23.  
    24. void *writer(void *args)
    25. {
    26.     const char *name = static_cast<const char *>(args);
    27.     sleep(1);
    28.     while(true)
    29.     {
    30.         pthread_rwlock_wrlock(&rw);
    31.         board++;
    32.         cout << "I am writer" << endl;
    33.         sleep(10);
    34.         pthread_rwlock_unlock(&rw);
    35.     }
    36. }
    37.  
    38. int main()
    39. {
    40.     pthread_rwlock_init(&rw, nullptr);
    41.     pthread_t r1,r2,r3,r4,r5,r6, w;
    42.     pthread_create(&r1, nullptr, reader, (void*)"reader");
    43.     pthread_create(&r2, nullptr, reader, (void*)"reader");
    44.     pthread_create(&r3, nullptr, reader, (void*)"reader");
    45.     pthread_create(&r4, nullptr, reader, (void*)"reader");
    46.     pthread_create(&r5, nullptr, reader, (void*)"reader");
    47.     pthread_create(&r6, nullptr, reader, (void*)"reader");
    48.     pthread_create(&w, nullptr, writer, (void*)"writer");
    49.  
    50.  
    51.     pthread_join(r1, nullptr);
    52.     pthread_join(r2, nullptr);
    53.     pthread_join(r3, nullptr);
    54.     pthread_join(r4, nullptr);
    55.     pthread_join(r5, nullptr);
    56.     pthread_join(r6, nullptr);
    57.     pthread_join(w, nullptr);
    58.  
    59.     pthread_rwlock_destroy(&rw);
    60.     return 0;
    61. }

  • 相关阅读:
    项目管理必看书籍
    第二章：Pythonocc官方demo 案例45（几何轴向曲线偏置）
    一个宁静祥和没有bug的下午和SqlSession的故事
    使用结构体组织相关联的数据(5)
    域名一定需要备案吗？什么情况下不用备案？
    SpringBoot2基础篇（三）—— 整合第三方技术
    一种适用于大量租户大量角色权限系统设计
    Swift 另辟蹊径极速生成图片的缩略图
    基于python的智慧城市社区服务平台及养老服务子系统的设计与实现
    摸鱼有理：大脑一思考就在积累毒素，必须休息才能清除｜Cell子刊
  • 原文地址：https://blog.csdn.net/zhang_si_hang/article/details/128033100