线程相关笔记

痛定思痛。 2021-12-11 08:35 309阅读 0赞

> 注意：本篇文章不是系统的线程知识讲解，更多的是一些实用的线程常识和个人的笔记总结。如果读者想要获取更系统的OS线程知识讲解，那么本篇文章并不能很好的对你带来帮助，请移步至更详细系统的OS线程相关教程。

## 一、简介 ##

--------------------

线程，有时又被称为轻量级进程，是程序执行流的最小单元。一个标准的线程由线程ID、当前指令指针（PC）、寄存器集合和堆栈组成。通常意义上，一个进程由一个到多个线程组成，各个线程之间共享程序的内存空间（包括代码段、数据段、堆等）及一些进程级的资源（如打开文件和信号）。一个经典的线程与进程的关系如下：

![线程和进程对应的关系模型][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3phbmRhXw_size_16_color_FFFFFF_t_70]

## 二、访问权限 ##

--------------------

线程的私有存储空间包括以下几个方面：

*  栈
 *  线程局部存储（TLS）。TLS是某些OS为线程单独提供的私有空间，通常容量很有限。
 *  寄存器（包括PC寄存器）

对于C/C++程序员，可以参照以下表格：

<table> 
 <thead> 
  <tr> 
   <th>线程私有</th> 
   <th>线程间共享</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>局部变量</td> 
   <td>全局变量</td> 
  </tr> 
  <tr> 
   <td>函数的参数</td> 
   <td>堆上数据</td> 
  </tr> 
  <tr> 
   <td>TLS数据</td> 
   <td>函数中的静态变量</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td>程序代码</td> 
  </tr> 
  <tr> 
   <td></td> 
   <td>文件描述符</td> 
  </tr> 
 </tbody> 
</table>

## 三、线程调度 ##

不论是在多处理器的计算机上还是在单处理器的计算机上，线程总是“并发”执行的。当线程数量小于等于处理器数量时（并且OS支持多处理器），线程的并发是真正的并发，即不同的线程运行在不同的处理器上，彼此之间互不相干。但对于线程数量大于处理器数量的情况，线程的并发就不是真正的并发，因为此时至少会有一个处理器运行多个线程，为了更清楚的描述这种情况，下面用单处理器-多线程来说明：

在单处理器对应多线程的情况下，这种并发，是一种模拟出来的状态。具体是这样的：OS会让这些多线程程序轮流执行，每次仅执行一小段时间（eg：几十到几百毫秒），这样每个线程看起来就像是在“同时”执行。

这样的一个不断在处理器上切换不同的线程的行为势必会造成一些资源的浪费（线程切换浪费的时间），所以对应高性能服务器开发的一个建议就是：**限制活跃线程数少于或等于CPU个数。**

> 这里对应的知识点参考：[提升服务器性能的建议:减少环境切换带来的开销][Link 1]

一般情况下，通常把很少陷入等待状态的线程称为 **CPU密集型线程**，而把频繁陷入等待状态的线程称为 **IO密集型线程。** IO密集型线程总是比CPU密集型线程更容易得到优先级的提升。（进程也类似）

## 四、线程安全 ##

--------------------

//TODO

## 五、优点 ##

--------------------

*  某个操作可能会陷入长时间等待，等待的线程会进入睡眠状态，无法继续执行。多线程执行可以有效利用等待的时间。eg：等待网络响应。
 *  某个操作会消耗大量的时间（eg：计算），如果只有一个线程，程序和用户的交互就会中断。多线程就可以让一个线程负责交互，一个线程负责计算。
 *  程序逻辑本身就要求并发操作。eg：多端下载、监听不同事件……
 *  多CPU或者多core计算机，本身具备同时执行多个线程的能力，因此单线程程序无法全面的发挥计算机的全部性能。
 *  多线程在数据共享方面相比多进程有天然的优势。

## 六、缺点 ##

*  健壮性不足。eg：一个线程出现异常退出，会导致整个进程退出。
 *  多线程编程的难度比较高，需要开发人员更严谨。eg：如果进程的其中一个线程出现 Segmentation falut 而进程中又没有SIGSEGV 信号的对应handler函数，那么默认的动作就会导致整个进程退出。

## 七、实例 ##

### 1. 生产者消费者模型 ###

> 生产者指的是生产数据的任务，消费者指的是处理数据的任务。在并发编程中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题，引入了生产者和消费者模式。

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

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

基于阻塞队列实现生产者消费者模型：

#include <iostream>
    #include <queue>
    #include <unistd.h>
    #include <pthread.h>
    
    #define NUM 8
    #define CONSUMER_COUNT 2
    #define PRODUCTER_COUNT 2
    
    class BlockQueue{ 
    private:
      std::queue<int> q;
      size_t cap;
      pthread_mutex_t lock;
      pthread_cond_t full;
      pthread_cond_t empty;
    private:
      void LockQueue(){ 
        pthread_mutex_lock(&lock);
      }
      void UnLockQueue(){ 
        pthread_mutex_unlock(&lock);
      }
      void ProductWait(){ 
        pthread_cond_wait(&full, &lock);
      }
      void ConsumerWait(){ 
        pthread_cond_wait(&empty, &lock);
      }
      void NotifyProduct(){ 
        pthread_cond_signal(&full);
      }
      void NotifyConsummer(){ 
        pthread_cond_signal(&empty);
      }
      bool IsEmpty(){ 
        return q.size() == 0;
      }
      bool IsFull(){ 
        return q.size() == cap;
      }
    public:
      BlockQueue(int _cap = NUM) : cap(_cap){ 
        pthread_mutex_init(&lock, NULL);
        pthread_cond_init(&full, NULL);
        pthread_cond_init(&empty, NULL);
      }
      void PushData(const int& data){ 
        LockQueue();
        while(IsFull()){ 
          NotifyConsummer();
          std::cout << "queue full, notify consumer. product stop!" << std::endl;
          ProductWait();
        }
        q.push(data);
        NotifyConsummer();
        UnLockQueue();
      }
      void PopData(int& data){ 
        LockQueue();
        while(IsEmpty()){ 
          NotifyProduct();
          std::cout << "queue empty, notify producter. consummer stop!" << std::endl;
          ConsumerWait();
        }
        data = q.front();
        q.pop();
        NotifyProduct();
        UnLockQueue();
      }
      ~BlockQueue(){ 
        pthread_mutex_destroy(&lock);
        pthread_cond_destroy(&full);
        pthread_cond_destroy(&empty);
      }
    };
    
    pthread_t threads[CONSUMER_COUNT + PRODUCTER_COUNT];
    
    void* consumer(void* arg){ 
      BlockQueue* b = (BlockQueue*)arg;
      int data;
      while(true){ 
        b->PopData(data);
        std::cout << "Consum data done: " << data << std::endl;
      }
    }
    
    void* producter(void* arg){ 
      BlockQueue* b = (BlockQueue*)arg;
      srand((unsigned long)time(NULL));
      while(true){ 
        int data = rand() % 1024;
        b->PushData(data);
        std::cout << "Product data done: " << data << std::endl;
        sleep(1);
      }
    }
    
    int main(){ 
      BlockQueue b;
      for(int i = 0; i != CONSUMER_COUNT; ++i){ 
          pthread_create(&threads[i], NULL, consumer, (void*)&b);
      }
      for(int i = 0; i != PRODUCTER_COUNT; ++i){ 
          pthread_create(&threads[CONSUMER_COUNT + i], producter, (void*)&b);
      }
      for(int i = 0; i != (CONSUMER_COUNT + PRODUCTER_COUNT); ++i){ 
          pthread_joib(threads[i], NULL);
      }
      return 0;
    }

### 2. 线程池 ###

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

线程池常见的应用场景：

*  需要大量的线程来完成任务，但是任务时间都比较短。典型的一个案例就是Web服务器完成网页请求。对于长时间的任务，比如telnet连接请求，因为telnet会话保持时间可能远大于线程的创建时间，所以创建线程的时间开销就可以忽略不计。
 *  对性能严格要求的应用。
 *  接收突发的大量请求，但不至于使服务器因此产生大量线程的应用。

一个高效且经典的线程池示例：

//代码来源：https://github.com/progschj/ThreadPool
    #ifndef THREAD_POOL_H
    #define THREAD_POOL_H
    
    #include <vector>
    #include <queue>
    #include <memory>
    #include <thread>
    #include <mutex>
    #include <condition_variable>
    #include <future>
    #include <functional>
    #include <stdexcept>
    
    class ThreadPool { 
    public:
        ThreadPool(size_t);
        template<class F, class... Args>
        auto enqueue(F&& f, Args&&... args) 
            -> std::future<typename std::result_of<F(Args...)>::type>;
        ~ThreadPool();
    private:
        // need to keep track of threads so we can join them
        std::vector< std::thread > workers;
        // the task queue
        std::queue< std::function<void()> > tasks;
        
        // synchronization
        std::mutex queue_mutex;
        std::condition_variable condition;
        bool stop;
    };
     
    // the constructor just launches some amount of workers
    inline ThreadPool::ThreadPool(size_t threads)
        :   stop(false)
    { 
        for(size_t i = 0;i<threads;++i)
            workers.emplace_back(
                [this]
                { 
                    for(;;)
                    { 
                        std::function<void()> task;
    
                        { 
                            std::unique_lock<std::mutex> lock(this->queue_mutex);
                            this->condition.wait(lock,
                                [this]{  return this->stop || !this->tasks.empty(); });
                            if(this->stop && this->tasks.empty())
                                return;
                            task = std::move(this->tasks.front());
                            this->tasks.pop();
                        }
    
                        task();
                    }
                }
            );
    }
    
    // add new work item to the pool
    template<class F, class... Args>
    auto ThreadPool::enqueue(F&& f, Args&&... args) 
        -> std::future<typename std::result_of<F(Args...)>::type>
    { 
        using return_type = typename std::result_of<F(Args...)>::type;
    
        auto task = std::make_shared< std::packaged_task<return_type()> >(
                std::bind(std::forward<F>(f), std::forward<Args>(args)...)
            );
            
        std::future<return_type> res = task->get_future();
        { 
            std::unique_lock<std::mutex> lock(queue_mutex);
    
            // don't allow enqueueing after stopping the pool
            if(stop)
                throw std::runtime_error("enqueue on stopped ThreadPool");
    
            tasks.emplace([task](){  (*task)(); });
        }
        condition.notify_one();
        return res;
    }
    
    // the destructor joins all threads
    inline ThreadPool::~ThreadPool()
    { 
        { 
            std::unique_lock<std::mutex> lock(queue_mutex);
            stop = true;
        }
        condition.notify_all();
        for(std::thread &worker: workers)
            worker.join();
    }
    
    #endif

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3phbmRhXw_size_16_color_FFFFFF_t_70]: /images/20211211/7f8e84085c9645ac95b69e8438599d73.png
[Link 1]: https://blog.csdn.net/zanda_/article/details/80156634