首页 > 其他分享 >Muduo库之线程

Muduo库之线程

时间：2022-10-16 16:00:09浏览次数：72  
标签：Muduo 函数 void _. 线程 key pthread

Thread

在 Thread.cc 中，有一个 ThreadNameInitializer 类，用于线程环境初始化操作：

void afterFork() {
  muduo::CurrentThread::t_cachedTid = 0;
  muduo::CurrentThread::t_threadName = "main";
  CurrentThread::tid();
}

class ThreadNameInitializer {
public:
  ThreadNameInitializer() {
    muduo::CurrentThread::t_threadName = "main";
    CurrentThread::tid();
    pthread_atfork(NULL, NULL, &afterFork);
  }
};

ThreadNameInitializer init;

在 Linux 系统开发中，如果父进程调用 fork() 函数派生子进程时，父进程同时也具有 pthread 的互斥锁对象，那么子进程将自动继承父进程中的互斥锁，且子进程中的互斥锁状态也会和父进程中的互斥锁状态保持一致，即如果父进程中的互斥锁处于加锁状态，那么子进程中的互斥锁也处于加锁状态；而如果父进程中的互斥锁处于未加锁状态，那么子进程中的互斥锁也处于未加锁状态。这就可能造成父进程和子进程可能对同一个资源进行加锁，形成死锁。为此，设计出了 pthread_atfork() 函数，其函数原型如下：

int pthread_atfork(void (*prepare)(void), void (*parent)(void), void (*child)(void));

其中：

  • prepare: 在父进程 fork 之前调用，这里可以获取父进程定义的所有锁变量；
  • parent: fork 创建子进程之后，但在 fork 返回之前的父进程中调用，在此处可以对 prepare 获得的锁进行解锁；
  • child: fork 返回之前在子进程中调用，在此处可以对 prepare 获得的锁进行解锁；

而 Muduo 中的 afterFork() 方法即是在 fork 返回之前，在子进程中进行调用，先对线程id号清零，然后由于子进程并不继承父进程的线程，所以此时子进程中只有一个线程，即为主线程，因此将 t_threadName 变量改为 main。最后调用 CurrentThread::tid() 方法重新获得子进程的id号。

有一个关键的结构体 ThreadData，用于存放线程相关的数据：

struct ThreadData
{
  typedef muduo::Thread::ThreadFunc ThreadFunc;
  ThreadFunc func_;
  string name_;
  pid_t* tid_;
  CountDownLatch* latch_;

  ThreadData(ThreadFunc func,
             const string& name,
             pid_t* tid,
             CountDownLatch* latch)
    : func_(std::move(func)),
      name_(name),
      tid_(tid),
      latch_(latch)
  { }
}

其内部有一个 runInThread() 方法，在指定的线程 id 中运行线程:

void runInThread() {
  *tid_ = muduo::CurrentThread::tid();
  tid_ = NULL;
  latch_->countDown();
  latch_ = NULL;

  muduo::CurrentThread::t_threadName = name_.empty() ? "muduoThread" : name_.c_str();
  ::prctl(PR_SET_NAME, muduo::CurrentThread::t_threadName);
  try {
    func_();
    muduo::CurrentThread::t_threadName = "finished";
  } catch (const Exception& ex) {
    muduo::CurrentThread::t_threadName = "crashed";
    fprintf(stderr, "exception caught in Thread %s\n", name_.c_str());
    fprintf(stderr, "reason: %s\n", ex.what());
    fprintf(stderr, "stack trace: %s\n", ex.stackTrace());
    abort();
  } catch (const std::exception& ex) {
    muduo::CurrentThread::t_threadName = "crashed";
    fprintf(stderr, "exception caught in Thread %s\n", name_.c_str());
    fprintf(stderr, "reason: %s\n", ex.what());
    abort();
  } catch (...) {
    muduo::CurrentThread::t_threadName = "crashed";
    fprintf(stderr, "unknown exception caught in Thread %s\n", name_.c_str());
    throw; // rethrow
  }
}

首先通过 CurrentThread::tid() 函数将当前线程的 id 缓存在 t_cachedTid 变量中。然后将 CountDownLatch 对象中的 count_ 数目减一，表示当前线程基本启动完成，直到其值为 0，表示所有子线程启动完成，则唤醒主线程。

随后更改线程名称，默认名称为 muduoThread。最后，运行真正的线程函数 func_，直到其运行完成，将其线程名称更改为 finished。在此期间，通过 Exception 异常类来捕获各种异常信息。

如下为新线程的入口函数：

void* startThread(void* obj) {
  ThreadData* data = static_cast<ThreadData*>(obj);
  data->runInThread();
  delete data;
  return NULL;
}

通过将万能指针强制转化为 ThreadData* 类型，以启动线程，阻塞在 runInThread() 函数处，直到线程工作完成，并清理相关资源。

成员变量 numCreated_ 则记录了当前所创建的线程数目，其类型为原子操作类型，节省了互斥锁资源，且相较于互斥锁则更为快速，具有线程安全。

static AtomicInt32 numCreated_;

主线程的启动函数如下：

void Thread::start() {
  assert(!started_);
  started_ = true;
  detail::ThreadData* data = new detail::ThreadData(func_, name_, &tid_, &latch_);
  if (pthread_create(&pthreadId_, NULL, &detail::startThread, data)) {
    started_ = false;
    delete data; // or no delete?
    LOG_SYSFATAL << "Failed in pthread_create";
  } else {
    latch_.wait();
    assert(tid_ > 0);
  }
}

此处通过 pthread_create() 函数创建新的线程并让该线程运行在 startThread() 函数中，如果创建成功，主线程就阻塞在 wait() 处，在子线程运行真正的线程函数之前，即调用 ThreadFunc 之前唤醒主线程。

调用链如下图所示：

ThreadLocal

在单线程程序中，可以使用 “全局变量” 在多个函数之间共享数据。然而，则多线程环境下，由于数据空间是共享的，所以全局变量也会被所以线程所拥有。但是有时有必要提供线程私有的全局变量，仅在某个线程中有效，因此，在 POSIX 线程库中，可以通过 线程特定数据(Thread-specific Data, TSD) 这个数据结构来解决该问题，它也成为 线程本地存储(Thread-local Storage)，而对于 POD 类型，可以通过 __thread 关键字来进行修饰，比如 CurrentThread.h 文件中的如下内容：

extern __thread int         t_cachedTid;
extern __thread char        t_tidString[32];
extern __thread int         t_tidStringLength;
extern __thread const char* t_threadName;

而 ThreadLocal 便是封装了该机制，在 POSIX 线程库中通常通过如下四种方法来实现该机制:

int pthread_key_create(pthread_key_t *key, void (*destructor)(void*));
int pthread_key_delete(pthread_key_t key);
void *pthread_getspecific(pthread_key_t key);
int pthread_setspecific(pthread_key_t key, const void *value);

POSIX 要求实现 POSIX 的系统为每个进程维护一个称为 Key 的结构数组，这个数组中每个结构称为一个线程特定数据元素，如图所示。POSIX 规定系统实现的 Key 结构数组必须包含不少于 128 个线程特定元素，而每个线程特定数据元素至少包含两个内容：使用标志和析构函数指针。其中的标志用于指示这个数组元素是否使用，所有标志初始化为 “未使用”。

当一个线程调用 pthread_key_create() 创建一个新线程特定数据元素时，系统搜索其所在进程的 Key 结构数组，找出其中第一个未使用的元素，并通过 keyptr 返回该元素的键，即数组索引。pthread_key_create() 函数的第二个参数是一个函数指针，指向一个析构函数，用于线程结束后一些内存清理工作，析构函数的参数就是线程特定数据的指针。

除了进程范围内的 Key 结构数组外，系统还在进程中维护关于每个线程的线程结构，把这个关于线程的特定结构称为 pthread结构。其部分内容是关于 Key 数组对应的指针数组，它的 128 个指针和进程中的 128 个可能的索引是逐一关联的，指针指向的内存就是线程特有数据。

具体的关联情况如下：

而另外三个函数，pthread_key_delete() 用于删除 key，而非删除数据，数据的清理需要通过之前所创建的回调函数来清理。pthread_getspecific() 用于返回线程特定数据，如果没有值进行关联会返回 NULL。prthread_setspecific() 则用于设置线程特定数据的值。

注意，不论哪个线程调用了 pthread_key_create()，所创建的 key 都是所有线程可以访问的，但各个线程可以根据自己的需要往 key 中填入不同的值，相当于提供了一个同名但不同值的全局变量。

ThreadLocal 类使用 RAII 方式，在构造函数中创建 key，在构造函数中删除 key：

ThreadLocal() {
  MCHECK(pthread_key_create(&pkey_, &ThreadLocal::destructor));
}

~ThreadLocal() {
  MCHECK(pthread_key_delete(pkey_));
}

其指定的回调函数如下:

static void destructor(void *x) {
  T* obj = static_cast<T*>(x);
  typedef char T_must_be_complete_type[sizeof(T) == 0 ? -1 : 1];
  T_must_be_complete_type dummy; (void) dummy;
  delete obj;
}

其中使用 typedef 关键字定义了一个数组类型，用于在编译期判断类型 T 是否是不完全类型，不完全类型指的是只有声明却没有定义的类，那么不完全类型在 delete 操作时也就无法调用析构函数，因此在 delete value_ 操作之前需要判断类型 T 是否为不完全类型。

如果是不完全类型，那么 sizeof(T) == 0 为真，数组大小为 -1，编译错误；如果是完全类型，那么 sizeof(T) == 0 为假，数组大小为 1，编译成功。

ThreadLocalSingleton

ThreadLocalSingleton 类封装为了线程本地存储单例类。在 Signleton 中提到过，常见的单例模式设计分为四种：

  • 懒汉式
  • 双检锁
  • 饿汉式
  • 局部静态式

ThreadLocalSingleton 类则采用了懒汉式的构建方法，而且由于是线程本地存储，则无需互斥锁来保证线程安全：

static T& instance() {
  if (!t_value_) {
    t_value_ = new T();
    deleter_.set(t_value_);
  }
  return *t_value_;
}

ThreadLocalSignleton 类有两个成员变量：

static __thread T* t_value_;
static Deleter deleter_;

其中，t_value_ 用于构建单例对象，由于 T* 为指针类型，即 POD 类型，因此可以使用 __thread 关键字进行修饰，使其成为线程本地存储数据，而无需使用 ThreadLocal 类型。其次，deleter_ 变量则用于声明一个 Deleter 类对象，该类用于销毁 t_value_ 所指的对象，其定义如下：

class Deleter {
public:
  Deleter() {
    pthread_key_create(&pkey_, &ThreadLocalSingleton::destructor);
  }

  ~Deleter() {
    pthread_key_delete(pkey_);
  }

  void set(T* newObj) {
    assert(pthread_getspecific(pkey_) == NULL);
    pthread_setspecific(pkey_, newObj);
  }

  pthread_key_t pkey_;
};

可见，Deleter 是一个 TSD 类型，只为线程本地所拥有。类似于 ThreadLocal 类型，它使用 RAII 方式，在构造函数中创建 key，以绑定 ThreadLocalSingleton 类中的销毁函数，而在析构函数中则删除 key。

销毁对象的函数则如下：

static void destructor(void* obj) {
  assert(obj == t_value_);
  typedef char T_must_be_complete_type[sizeof(T) == 0 ? -1 : 1];
  T_must_be_complete_type dummy; (void) dummy;
  delete t_value_;
  t_value_ = 0;
}

同样的，首先判断类型 T 是否为不完全类型，如果是，则会在编译期发生错误，否则可以调用 delete 以调用析构函数，销毁该对象。

ThreadPool

线程池就是提前创建若干个线程，如果有任务需要处理，线程池里的线程就会处理任务，处理完之后线程并不会被销毁，而是等待下一个任务。由于创建和销毁线程都是消耗系统资源的，所以当去频繁的创建和销毁线程的时候就可以考虑采用线程池来提升系统的性能。

Java 中线程池的设计比较完善，下面为 Java 中几种常见的线程池实现方式：

  • FixedThreadPool: 有限线程数的线程池。其特点是它的核心线程数和最大线程数是一样的，即固定线程数的线程池，因为它不会将超出线程数的线程缓存到队列中，如果超出线程数了，那么会按拒绝策略来执行。
  • CachedThreadPool: 无限线程数的线程池。这是一个可以缓存线程任务的线程池，并且直接执行，其特点是可以无限的缓存线程任务。除此之外，对于空闲线程还设置了 60 秒的等待时间，即如果线程在工作完成后，如果 60 秒内没有任务再进来，该线程就会被销毁，如果有，就继续使用，这样就在最大程度上保证了线程的灵活性。
  • ScheduledThreadPool: 定时线程池。这个线程池就是为了定时而发明的，它支持定时或周期性执行任务，比如 10 秒执行一次任务。
  • SingleThreadExecutor: 单一线程池。这个线程池适用于需要按照提交顺序去执行线程的场景，因为在该线程池中，只有一个线程可以执行，其原理类似于 FixedThreadPool，只不过其核心线程数和最大线程数都是 1，这样当提交者去提交线程时，就必须先让线程池中的线程执行完成后才会去执行接下来的线程。这样就保证了线程的顺序性。
  • SingleThreadScheduledExecutor: 单一定时线程池。该线程池即具有 SingleThreadExecutor 单一线程池的特性，一次执行一个线程，又具有 ScheduledThreadPool 线程池的定时功能。
  • ForkJoinPool: 孕妇线程池。该线程池有自己的一个公共队列，当这个公共队列执行的线程任务所创建出来的子线程任务将不会放到公共队列中，而是放到自己的单独队列中，这样就不会互相影响。减少了线程间的竞争和切换，提高了效率。当某一个线程的子任务很繁重，而另一个子线程的任务很少时，它通过双端队列来平衡各个线程之间的负载。

而 Muduo 中的线程池的设计就是一个固定大小的线程池。在 ThreadPool 类内有如下两个成员变量，分别用来存储线程和任务队列，线程容器预备有一定数量的线程备用，任务容器在有新的任务进入时，唤醒一个线程执行任务：

typedef std::function<void ()> Task;
std::vector<std::unique_ptr<muduo::Thread>> threads_;
std::deque<Task> queue_ GUARDED_BY(mutex_);

如下方法是为了将要执行的任务放入任务队列：

void ThreadPool::run(Task task) {
  if (threads_.empty()) {
    task();
  } else {
    MutexLockGuard lock(mutex_);
    while (isFull() && running_) {
      notFull_.wait();
    }
    if (!running_) return;
    assert(!isFull());

    queue_.push_back(std::move(task));
    notEmpty_.notify();
  }
}

首先判断线程池是否为空，如果为空，则说明线程池并未分配线程，那么就交给当前线程来执行任务。之后，如果任务队列已满，就一直阻塞，直到 notFull_ 条件变量唤醒该线程。最后将新的任务放入任务队列的尾端，并通过 notEmpty_ 条件变量唤醒某个等待取任务的线程进行工作。

而从队列中取出任务的方法如下：

ThreadPool::Task ThreadPool::take() {
  MutexLockGuard lock(mutex_);
  while (queue_.empty() && running_) {
    notEmpty_.wait();
  }
  Task task;
  if (!queue_.empty()) {
    task = queue_.front();
    queue_.pop_front();
    if (maxQueueSize_ > 0) {
      notFull_.notify();
    }
  }
  return task;
}

首先判断任务队列是否为空，如果为空，则一直阻塞，直到 notEmpty_ 条件变量唤醒该线程。然后从任务队列的头部取出任务元素，并通过 notFull_ 条件变量唤醒某个等待放任务的线程进行工作。

线程池的启、停方法如下：

void ThreadPool::start(int numThreads) {
  assert(threads_.empty());
  running_ = true;
  threads_.reserve(numThreads);
  for (int i = 0; i < numThreads; ++i) {
    char id[32];
    snprintf(id, sizeof id, "%d", i+1);
    threads_.emplace_back(new muduo::Thread(
          std::bind(&ThreadPool::runInThread, this), name_+id));
    threads_[i]->start();
  }
  if (numThreads == 0 && threadInitCallback_) {
    threadInitCallback_();
  }
}

void ThreadPool::stop() {
  {
    MutexLockGuard lock(mutex_);
    running_ = false;
    notEmpty_.notifyAll();
    notFull_.notifyAll();
  }
  for (auto& thr : threads_) {
    thr->join();
  }
}

在启动函数中，首先对线程容器调用 reserve() 方法，使得该容器至少容纳 numThreads 个元素。然后循环创建线程，每个线程需要传递两个参数，第一个参数是线程的运行函数的函数指针，第二个参数则是线程名称。其中，所绑定的函数为 ThreadPool::runInThread()。完成之后，调用 Thread::start() 方法以真正的创建线程并启动。而如果线程数为 0，那么只需初始化一个线程即可。

在线程池停止方法中，通过条件变量唤醒所以线程，执行 join() 方法，退出所有线程并回收资源，防止资源泄漏。

注意，启动线程池时无需加锁，因为此时只是创建空线程，并没有具体的任务来进行调度，故为单线程，不存在数据被多线程改变的情况。而在停止线程池时，有共享数据的修改，所以需要加锁。

void ThreadPool::runInThread() {
  try {
    if (threadInitCallback_) {
      threadInitCallback_();
    }
    while (running_) {
      Task task(take());
      if (task) {
        task();
      }
    }
  }
  catch {
    // ...
  }
}

其中，先调用 threadInitCallback_() 回调函数进行线程初始化操作，然后从任务队列中获取任务，如果队列为空，则一直阻塞，直到任务队列中有任务出现，取出该任务并执行。如果有异常出现则抛出异常。

如图，为 Muduo 中线程池的简略图：

标签：Muduo,函数,void,_.,线程,key,pthread
From： https://www.cnblogs.com/tuilk/p/16796365.html

相关文章

  • GO语言sync包与锁：限制线程对变量的访问
    原文：http://c.biancheng.net/view/4307.htmlGo语言中sync包里提供了互斥锁Mutex和读写锁RWMutex用于处理并发过程中可能出现同时两个或多个协程（或线程）读或写同一个......
  • 多线程并行与分布式执行
    假如项目中有测试用例1000条，一条测试用例需要执行1分钟，一个测试人员需要1000分钟才能完成一个轮回测试。通常会用人力成本换取时间成本，加几个人的一起执行，时间就会缩短。如......
  • 线程，线程池和锁
    线程相关的面试题线程线程和进程有什么区别和联系？联系：进程是线程的【上级】和【容器】，一个进程中可以有一个或多个线程（至少一个）。线程概念是进程概念的轻量化，很多线......
  • 面试题：JAVA多线程交替打印ABC
    JAVA实现，3个线程交替A,B,C，一共完成10次“ABC”打印，结束后打印“END”。打印示例：abcabcabcabcabcabcabcabcabcabcEND 分析：打印10次ABC，3个线程分别打印A,B......
  • 2022.10.15线程状态观测
    线程状态观测线程优先级packagecom.fei.stata;​publicclassTestPriorityextendsThread{  publicstaticvoidmain(String[]args){    System.out......
  • 高并发和多线程
    高并发是一种状态。如果大量的访问网络中的某个接口，就会出现大量的操作，如数据库操作，资源的请求，硬件的占用等。这就需要对接口进行优化，对sql语句进行优化，而多线程是处理高并......
  • 线程创建的方式
    说说线程的创建方式继承thread类，重写run（）方法；实现Runnable接口，实现run()方法；实现callable接口，实现call方法；​ 使用Runnable接口和Callable接口的方式,区别在于前......
  • Muduo库之阻塞队列
    在并发编程中，常常需要用到线程安全的队列。常见的线程安全队列的设计分为两种：阻塞队列：常用于生产者和消费者的场景，其中，生产者存放元素，而消费者获取元素。常用的实现方法......
  • Muduo库之异步日志
    该框架中的日志为诊断日志，用于将代码运行时的重要信息进行保存，方便故障诊断和追踪。日志通常分为如下两种：同步日志：当需要写出一条日志消息时，只有等到这条日志消息完全写......
  • 动态线程池框架 hippo4j 源码解析
    hippo4j是一个动态管理和监控线程池的开源框架，它有两种运行模式：轻量级依赖配置中心以及无中间件依赖版本。文档地址参见https://hippo4j.cn/docs/user_docs/intro其中......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99