1. 封装线程类

来自thread.h、thread.cpp文件

1.1 核心逻辑

线程开发 - 图1

1.2 成员变量

  1. class Thread
  2. {
  3.     ...
  4.     ...
  5. private:
  6.     ///线程ID  用户态的线程ID和内核线程ID不是一个概念 调试时候需要拿到内核中的ID
  7.     pid_t m_id = -1;
  8.     ///线程号
  9.     pthread_t m_thread = 0;   //unsigned long int
  10.     ///线程执行的回调函数
  11.     std::function<void()> m_cb;
  12.     ///线程名称
  13.     std::string m_name;
  14.     //信号量对象
  15.     Semaphore m_sem;
  17. };

· 补充说明：开始定义线程局部变量

目的：存储管理当前线程的类对象Thread的this指针 和 当前线程的命名
（thread_local是C++11引入的类型符）
image.png

1.3 接口

1.3.1 构造函数

功能：利用pthread库开启运行线程，并且置一个信号量去等待线程完全开启后再退出构造函数

  1. Thread::Thread(std::function<void()> cb, const std::string& name)
  2.     :m_cb(cb)
  3. {
  4.     m_name = name.size() > 0 ? name : "UNKNOW";
  6.     //创建线程
  7.     int ret = pthread_create(&m_thread, nullptr, &Thread::run, this);
  8.     if(ret < 0)
  9.     {
  10.         KIT_LOG_ERROR(g_logger) << "Thread:: pthread_create fail, ret= " << ret << " thread name= " << m_name;
  12.         throw std::logic_error("pthread_create fail!!");
  13.     }
  15.     //试想 当执行完上面的API线程可能不会马上运行 手动等待一下直到线程完全开始运行
  16.     m_sem.wait();
  18. }

· 补充说明：线程类对象之间应禁止拷贝赋值

原理：使用C++11特性，在成员函数尾部添加=delete表示该成员函数被禁用。发生相应可调用场合会自动屏蔽函数调用。
image.png

1.3.2 析构函数

  1. /**
  2.  * @brief 线程类析构函数
  3.  */
  4. ~Thread();
  6. Thread:: ~Thread()
  7. {
  8.     //析构时候不马上杀死线程 而是将其置为分离态
  9.     if(m_thread) 
  10.         pthread_detach(m_thread);
  12. }

** 私有接口

· run()

功能：pthread库指定的线程格式，用于作为线程回调函数。其中包含线程真正需要去执行的函数体，以及对线程的一系列初始化。

注意：必须声明为static静态函数。

  2. /**
  3.  * @brief 线程回调函数（固定格式）
  4.  * @return void* 
  5.  */
  6. static void *run(void *);
  8. void *Thread::run(void * arg)
  9. {
  10.     //接收传入的this指针 因为是个static方法 因此依靠传入this指针很重要
  11.     Thread* thread = (Thread*)arg;
  13.     //初始化线程局部变量
  14.     t_thread = thread;
  15.     t_thread_name = thread->m_name;
  17.     //获取内核线程ID
  18.     thread->m_id = GetThreadId();  //util.cpp定义的方法
  20.     //给内核线程命名 但是只能接收小于等于16个字符
  21.     pthread_setname_np(pthread_self(), thread->m_name.substr(0, 15).c_str());
  23.     //防止m_cb中智能指针的引用不被释放 交换一次会减少引用次数
  24.     std::function<void()> cb;
  25.     cb.swap(thread->m_cb);  
  27.     //确保线程已经完全初始化好了 再进行唤醒
  28.     thread->m_sem.notify();
  30.     //调取执行函数
  31.     cb();
  33.     return 0;
  34. }

· 小技巧：在线程回调函数里面顺利使用类成员

原理：在调用pthread_create()线程API的时候，传入参数this指针，就能通过这个指针去访问到类中的成员，否则类成员静态方法中无法使用非静态成员变量。或者，另外写一个普通成员函数，在线程回调函数中去调用那个函数也能解决该问题。

  • **Thread::Thread()**中：

image.png

  • **void *Thread::run(void * arg)**线程回调函数中：

image.png

1.3.3 join()

功能：以阻塞态的形式回收处理线程资源

  1. /**
  2.  * @brief 线程回收处理函数
  3.  */
  4. void join();
  6. void Thread::join()
  7. {
  8.     if(m_thread)
  9.     {
  10.         int ret = pthread_join(m_thread, nullptr);
  11.         if(ret < 0)
  12.         {
  13.             KIT_LOG_ERROR(g_logger) << "Thread:: pthread_join fail, ret= " << ret << " thread name= " << m_name;
  15.             throw std::logic_error("pthread_join fail!!");
  16.         }
  18.         m_thread = 0;
  19.     }  
  21. }

1.3.4 其他常用接口

  1. /**
  2.  * @brief 获取线程ID
  3.  * @return pid_t 
  4.  */
  5. pid_t getId() const {return m_id;}
  7. /**
  8.  * @brief 获取线程名称
  9.  * @return const std::string& 
  10.  */
  11. const std::string& getName() const {return m_name;}
  14. /**
  15.  * @brief 获取当前正在运行的线程指针
  16.  * @return Thread* 
  17.  */
  18. static Thread* GetThis();
  19. Thread* Thread::GetThis()
  20. {
  21.     return t_thread;
  22. }
  24. /**
  25.  * @brief 获取当前运行线程的名称
  26.  * @return const std::string& 
  27.  */
  28. static const std::string& GetName();
  29. const std::string& Thread::GetName()
  30. {
  31.     return t_thread_name;
  32. }
  34. /**
  35.  * @brief 设置当前运行线程的名称 
  36.  * @param[in] name 线程名称 
  37.  */
  38. static void SetName(const std::string& name);
  39. void Thread::SetName(const std::string& name)
  40. {
  41.     if(t_thread)
  42.     {
  43.         t_thread->m_name = name;
  44.     }
  46.     t_thread_name = name;
  47. }

C++知识补充复习1：线程局部变量（TLS thread-local-storage）

出处：《深入理解C++11新特性解析与应用》P215
概念：有线程生命周期以及线程可见性的变量。
特点：线程创建开始时，创建变量；线程结束时自动释放。

解决的问题：全局变量/静态变量在多线程环境下的共享问题。并非所有情况下，这种类型的变量都需要进行存储空间的共享。例如POSIX标准下的错误码全局变量errno，我们希望在某个线程下错误码是全局生效的，但是又不希望这个错误码会被其他线程所影响，因此引出线程局部变量，使得每个线程在”全局”上可以定义一个变量，但是空间的存储不是和所有线程共享。

本质上：由于线程是共享同一进程下的资源，因此线程局部变量的存储空间也在所依附的进程下。其他线程是能看到这一块空间的，但是会有线程局部特性的函数限制其他线程对该空间的访问。

C++知识补充复习2：区域锁（Scoped Locking）

概念：区域锁（Scoped Locking）并非一种锁的种类，而是使用锁的一种模式。这种概念是C++中RAII（Resource Acquisition Is Initialization）模式具体体现之一，”资源需要即初始化”。基本思想：C++中的资源应该交由对象管理，资源在对象的构造函数中完成初始化，在对象的析构函数中被释放。
拿现在锁的这个例子，一个管理锁的对象在构造函数中完成对锁的初始化，在析构函数中完成对锁的资源释放。

2. 封装互斥锁、信号量

来自mutex.h、mutex.cpp文件
思想：运用C++中RAII的模式思想，完成由对象管理锁、信号量资源。

2.1 信号量代码如下：

  1. /**
  2.  * @brief 信号量类
  3.  */
  4. class Semaphore : Noncopyable
  5. {
  6. public:
  7.     /**
  8.      * @brief 信号量类构造函数
  9.      * @param[in] count 计数基准值, 一般以0位基准
  10.      */
  11.     Semaphore(uint32_t count = 0);
  13.     /**
  14.      * @brief 信号量类析构函数
  15.      */
  16.     ~Semaphore();
  18.     //数-1
  19.     void wait();
  20.     //数+1
  21.     void notify();
  23. private:
  24.     //禁止拷贝赋值
  25.     Semaphore(const Semaphore &) = delete;
  26.     Semaphore(const Semaphore &&) = delete;
  27.     Semaphore& operator=(const Semaphore &) = delete;
  29. private:
  30.     /// 信号量结构体
  31.     sem_t m_semaphore;
  32. };
  34. Semaphore::Semaphore(uint32_t count)
  35. {
  36.     //初始化信号量
  37.     if(sem_init(&m_semaphore, 0, count) < 0)
  38.     {
  39.         KIT_LOG_ERROR(g_logger) << "Semaphore::Semaphore sem_init fail";
  41.         throw std::logic_error("sem_init fail!!");
  42.     }
  43. }
  45. Semaphore::~Semaphore()
  46. {
  47.     //销毁信号量
  48.     sem_destroy(&m_semaphore);
  49. }
  51. //数-1
  52. void Semaphore::wait()
  53. {
  54.     while(1)
  55.     {
  56.         //递减1 函数成功时返回0; 出错返回-1
  57.         if(!sem_wait(&m_semaphore))
  58.             break;
  60.         //如果中断就继续 
  61.         if(errno == EINTR)
  62.             continue;
  64.         KIT_LOG_ERROR(g_logger)  << "Semaphore::wait sem_wait error";
  65.         throw std::logic_error("sem_wait error!!");
  67.     }
  68. }
  70. //数+1
  71. void Semaphore::notify()
  72. {
  73.     //递增1 函数成功时返回0; 出错返回-1
  74.     if(sem_post(&m_semaphore) < 0)
  75.     {
  76.         KIT_LOG_ERROR(g_logger) << "Semaphore::notify sem_post error";
  78.         throw std::logic_error("sem_post error!!");
  79.     }
  81. }

2.2 互斥锁代码如下：

  1. /**
  2.  * @brief 局部区域互斥锁模板类
  3.  * @tparam T 
  4.  */
  5. template<class T>
  6. class ScopedLockImpl
  7. {
  8. public:
  9.     /**
  10.      * @brief 构造时加锁
  11.      * @param[in] mutex 互斥锁变量 
  12.      */
  13.     ScopedLockImpl(T &mutex)
  14.         :m_mutex(mutex)
  15.     {
  16.         m_mutex.lock();
  17.         m_locked = true;
  18.     }
  20.     /**
  21.      * @brief 析构时解锁
  22.      */
  23.     ~ScopedLockImpl()
  24.     {
  25.         unlock();
  26.     }
  28.     void lock()
  29.     {
  30.         if(!m_locked)
  31.         {
  32.             m_locked = true;
  33.             m_mutex.lock();
  34.         }
  35.     }
  37.     void unlock()
  38.     {
  39.         if(m_locked)
  40.         {
  41.             m_mutex.unlock();
  42.             m_locked = false;
  43.         }
  44.     }
  46. private:
  47.     /// 锁资源
  48.     T& m_mutex;
  49.     /// 上锁状态
  50.     bool m_locked;
  52. };
  55. /**
  56.  * @brief 互斥锁类
  57.  */
  58. class Mutex : Noncopyable
  59. {
  60. public:
  61.     typedef ScopedLockImpl<Mutex> Lock;
  63.     /**
  64.      * @brief 互斥锁类构造函数
  65.      */
  66.     Mutex() {pthread_mutex_init(&m_mutex, nullptr);}
  68.     /**
  69.      * @brief 互斥锁类析构函数
  70.      */
  71.     ~Mutex() {pthread_mutex_destroy(&m_mutex);}
  73.     /**
  74.      * @brief 加锁
  75.      */
  76.     void lock() {pthread_mutex_lock(&m_mutex);}
  78.     /**
  79.      * @brief 解锁
  80.      */
  81.     void unlock() {pthread_mutex_unlock(&m_mutex);}
  82. private:
  83.     /// 互斥锁
  84.     pthread_mutex_t m_mutex;
  86. };

2.3 读写锁代码如下：

  1. /**
  2.  * @brief 局部区域读锁模板类
  3.  * @tparam T 锁类型
  4.  */
  5. template<class T>
  6. class ReadScopedLockImpl
  7. {
  8. public:
  9.     ReadScopedLockImpl(T &mutex)
  10.         :m_mutex(mutex)
  11.     {
  12.         m_mutex.rdlock();
  13.         m_locked = true;
  14.     }
  16.     ~ReadScopedLockImpl()
  17.     {
  18.         unlock();
  19.     }
  21.     void lock()
  22.     {
  23.         if(!m_locked)
  24.         {
  25.             m_locked = true;
  26.             m_mutex.rdlock();
  27.         }
  28.     }
  30.     void unlock()
  31.     {
  32.         if(m_locked)
  33.         {
  34.             m_mutex.unlock();
  35.             m_locked = false;
  36.         }
  37.     }
  39. private:
  40.     T& m_mutex;
  41.     bool m_locked;
  43. };
  45. /**
  46.  * @brief 局部区域写锁模板类
  47.  * @tparam T 锁类型
  48.  */
  49. template<class T>
  50. class WriteScopedLockImpl
  51. {
  52. public:
  53.     WriteScopedLockImpl(T &mutex)
  54.         :m_mutex(mutex)
  55.     {
  56.         m_mutex.wrlock();
  57.         m_locked = true;
  58.     }
  60.     ~WriteScopedLockImpl()
  61.     {
  62.         unlock();
  63.     }
  65.     void lock()
  66.     {
  67.         if(!m_locked)
  68.         {
  69.             m_locked = true;
  70.             m_mutex.wrlock();
  71.         }
  72.     }
  74.     void unlock()
  75.     {
  76.         if(m_locked)
  77.         {
  78.             m_mutex.unlock();
  79.             m_locked = false;
  80.         }
  81.     }
  83. private:
  84.     T& m_mutex;
  85.     bool m_locked;
  87. };
  89. /**
  90.  * @brief 读写锁类
  91.  */
  92. class RWMutex: Noncopyable
  93. {
  94. public:
  95.     typedef ReadScopedLockImpl<RWMutex> ReadLock;
  96.     typedef WriteScopedLockImpl<RWMutex> WriteLock;
  98.     /**
  99.      * @brief 读写锁类构造函数 初始化读写锁
  100.      */
  101.     RWMutex() {pthread_rwlock_init(&m_rwlock, nullptr);}
  103.     /**
  104.      * @brief 读写锁类构造函数 初始化读写锁
  105.      */
  106.     ~RWMutex() {pthread_rwlock_destroy(&m_rwlock);}
  108.     /**
  109.      * @brief 读加锁
  110.      */
  111.     void rdlock() {pthread_rwlock_rdlock(&m_rwlock);}
  113.     /**
  114.      * @brief 写加锁
  115.      */
  116.     void wrlock() {pthread_rwlock_wrlock(&m_rwlock);}
  118.     /**
  119.      * @brief 读写全部解锁
  120.      */
  121.     void unlock() {pthread_rwlock_unlock(&m_rwlock);}
  123. private:
  124.     /// 读写锁
  125.     pthread_rwlock_t m_rwlock;
  126. };

2.4 自旋锁代码如下：

  1. **
  2.  * @brief 自旋锁类
  3.  */
  4. class SpinMutex: Noncopyable
  5. {
  6. public:
  7.     typedef ScopedLockImpl<SpinMutex> Lock;
  9.     /**
  10.      * @brief 自旋锁类构造函数 
  11.      */
  12.     SpinMutex() {pthread_spin_init(&m_mutex, 0);}
  14.     /**
  15.      * @brief 自旋锁类析构函数
  16.      */
  17.     ~SpinMutex() {pthread_spin_destroy(&m_mutex);}
  19.     /**
  20.      * @brief 加锁
  21.      */
  22.     void lock() {pthread_spin_lock(&m_mutex);}
  24.     /**
  25.      * @brief 解锁
  26.      */
  27.     void unlock() {pthread_spin_unlock(&m_mutex);}
  29. private:
  30.     /// 自旋锁
  31.     pthread_spinlock_t m_mutex;
  32. };

2.5 原子操作代码如下：

  1. **
  2.  * @brief 原子锁类
  3.  */
  4. class CASMutex: Noncopyable
  5. {
  6. public:
  7.     typedef ScopedLockImpl<CASMutex> Lock;
  9.     /**
  10.      * @brief 原子锁类构造函数
  11.      */
  12.     CASMutex(){ m_mutex.clear();}
  14.     /**
  15.      * @brief 原子锁类析构函数
  16.      */
  17.     ~CASMutex(){}
  19.     /**
  20.      * @brief 加锁
  21.      */
  22.     void lock() 
  23.     {
  24.         //执行本次原子操作之前 所有读原子操作必须全部完成
  25.         while(std::atomic_flag_test_and_set_explicit(&m_mutex, std::memory_order_acquire));
  26.     }
  28.     /**
  29.      * @brief 解锁
  30.      */
  31.     void unlock() 
  32.     {
  33.         //执行本次原子操作之前 所有写原子操作必须全部完成
  34.         std::atomic_flag_clear_explicit(&m_mutex, std::memory_order_release);
  35.     }
  38. private:
  39.     /// 原子锁
  40.     volatile std::atomic_flag m_mutex;
  42. };

C++知识补充复习3：信号量semaphore

概念：信号量是一种特殊的变量，对其操作访问都是源原子操作。且只允许等待wait和发送post操作

作用：通常用于保护共享资源，保证资源在任何时刻只有一个线程（进程）在访问。

线程开发 - 图6

主要使用的是POSIX 信号量，定义在头文件<semaphore.h>, 底层是一个非负整数，通过原子操作对其加减，控制线程（进程）对共享资源的访问；SYSTEM V 信号量定义在头文件<sys/sem.h>，底层是一个信号量集合，同样通过原子操作管理集合，但是操作比较复杂。

  • POSIX 信号量API:

    • int sem_init(sem_t *sem, int pshared, unsigned int value); 初始化——-构造
    • int sem_wait(sem_t *sem); 减1操作———-wait
    • int sem_post(sem_t *sem); 加1操作———-notify
    • int sem_destroy(sem_t *sem); 释放———-析构

  • SYSTEM V 信号量API: ```cpp union semun //信号量集合 系统定义 { int val; struct semid_ds buf; unsigned short arry; };

struct sembuf //信号量属性集合 系统定义好的 { unsigned short sem_num; //非负计数 short sem_op; //操作形式 加/减 short sem_flg; //模式 };

int semget(key_t key, int nsems, int semflg); //创建或引用一个信号量集合 int semctl(int semid, int semnum, int cmd, …); //对信号量集合指定操作 int semop(int semid, struct sembuf *sops, size_t nsops); //修改并自动执行信号量的操作指令 ```

  • 信号量和互斥锁之间的区别：

① 概念上，互斥锁只保证线程间对一个资源的互斥访问；信号量不仅保证互斥，还能控制线程同步
可以实现多个同类资源的多线程有序访问。
②用法上，互斥锁的加锁和解锁必须由同一个线程完成；信号量可以一个线程释放，另一个线程得到