其他分享
首页 > 其他分享> > 【muduo/base】线程

【muduo/base】线程

作者:互联网

#前言
muduo库也封装了线程Thread类,线程池ThreadPool。以及线程辅助类ThreadData,在这一部分我们可以学到如何创建线程,如何回收线程,加深对进程和线程的理解,以及代码规范方面的提升。

Thread类

头文件代码(去除了和命名相关的成员变量和函数)

class Thread : noncopyable
{
 public:
  typedef std::function<void ()> ThreadFunc;

  explicit Thread(ThreadFunc);
  ~Thread();
  void start();
  int join();
  bool started() const { return started_; }
  pid_t tid() const { return tid_; }
  const string& name() const { return name_; }
  static int numCreated() { return numCreated_.get(); }
  
 private:
 
  bool       started_;
  bool       joined_;
  pthread_t  pthreadId_;
  pid_t      tid_;
  ThreadFunc func_;
  CountDownLatch latch_;

};

这里我们归纳一下成员变量作用

线程状态相关: started_, joined_
线程信息相关 : pthreadId_,tid_
func_: 传入函数对象
代码执行相关 : latch_

到这里我们可能会疑惑:什么是tid?tid和threadid的区别是什么?

tid和threadid区别

在这里我们可以顺便复习一下操作系统相关知识
linux进程创建详细过程
进程和线程以及线程组
pid,tid,tgId区别

我们知道linux创建进程是通过(fork)分裂+替换(execv)生成新的进程。但是使用fork时,不知道你有没有考虑过下面这种情况:

进程A 创建了线程a,然后执行fork出了子进程B。那么这时候A创建的线程a,是否也会复制到进程B?
答案是不会?

这时我们是否会重新思考一下进程到底是什么?或许进程和线程不是一个包含关系?

这里做一下总结

  • 操作系统对进程的和线程调度算法并没有区别。进程和线程是地位相同的调度实体,在内核看来,他们都有一个统一的名字lwp(轻量级线程)
  • 线程组:是包含了多个lwp的集合,线程组id(tgid)为进程id,也就第一个lwp的pid
  • 进程组: 因为进程和线程对调度来看都一样,那么自然有进程组,进程组id(pgid)为父进程的pid
  • 我们使用ps打印的进程id,即线程组中第一个创建的lwp的pid,也即线程该lwp所在的线程组id
  • tid : tid就是lwp 的pid ,整个操作系统唯一
  • 线程id: 线程id是一个线程组内的唯一id 可以thread_self()获得。

typedef 和访问控制的关系

这里还有值得学习的地方,如typedef的位置
typedef定义为public ,用户可以直接使用Thread::ThreadFunc;定义为private,则仅可成员函数使用;定义为protected,则子类也可以使用

源文件代码

核心代码(删减非必要逻辑)

	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)
      {
      }

      void runInThread()
      {
        *tid_ = muduo::CurrentThread::tid();
        tid_ = NULL;
        latch_->countDown();
        latch_ = NULL;
        
        try
        {
          func_();
        }
        catch (...)
        {
          throw; // rethrow
        }
      }
    };
    void *startThread(void *obj)
    {
      ThreadData *data = static_cast<ThreadData *>(obj); //向下转型
      data->runInThread();
      delete data;
      return NULL;
    }
  Thread::Thread(ThreadFunc func)
      : started_(false),
        joined_(false),
        func_(std::move(func)),
        latch_(1) 
  {

  }

  Thread::~Thread()
  {
    if (started_ && !joined_)
    {
      pthread_detach(pthreadId_);
    }
  }


  void Thread::start()
  {
    assert(!started_);
    started_ = true;
    // FIXME: move(func_)
    detail::ThreadData *data = new detail::ThreadData(func_, name_, &tid_, &latch_); // data这里是用来做传出参数的,相当于一个代理,更新thread的tid以及,取消start的堵塞
    if (pthread_create(&pthreadId_, NULL, &detail::startThread, data))               //第三个参数为startThread()线程任务,传入参数void*
    {
      //失败
      started_ = false;
      delete data; // or no delete?
      LOG_SYSFATAL << "Failed in pthread_create";
    }
    else
    {
      // 成功
      latch_.wait();
      assert(tid_ > 0); //线程运行后,这里阻塞解除
    }
  }

  int Thread::join()
  {
    assert(started_);
    assert(!joined_);
    joined_ = true;
    return pthread_join(pthreadId_, NULL);
  }

ThreadData类

该类是一个辅助类,可以实现参数传入以及参数传出的功能,是一个代理类。创建线程的系统调用pthread_create,只能给线程绑定函数传入一个参数因此,需要一个额外类做参数包裹。

std::move

该函数功能可以将一个左值变为右值,如果某个类声明了右值构造函数(c++primer中有详细介绍),传参时不会经历对象的拷贝,而是将传入对象的控制权转移给新对象,并且将传入对象置为可析构状态,在适当情况下回收。使用右值引用需要慎重,如果传入对象被其他地方引用,可能会导致段错误。

代码规范

我们可以把只在当前源文件使用的类,定义在该源文件中

Thread类

一个执行的过程

创建thread对象threadA => threadA->start() => 创建threadData对象dataA => 系统调用创建线程,并绑定函数startThread => 创建线程成功 => 代码堵塞在 latch_.wait()处 => 线程startThread正式运行起来 => 调用传入参数dataA ->runInThread()方法 => 使threadA->latch_->countDown() => 主线程latch_.wait()检测到count<0,解除堵塞,返回执行调用threadA->start()之后的逻辑。 =》 主线程threadA->join() 堵塞在该处,直到线程任务结束线程资源回收完。

如果忘记调用了Join怎么办?

该类在Thread析构函数中调用了系统调用pthread_detach。该系统调用的作用是把该线程状态标记为detached状态,待线程执行完自动回收资源。对Thread对象销毁,但线程未被进行回收这种异常做了处理

  Thread::~Thread()
  {
    if (started_ && !joined_)
    {
      pthread_detach(pthreadId_);
    }
  }

分析

析构中调用pthread_detach将pthread_create状态置为detached状态,线程结束后,会自动被系统回收

ThreadPool类

设计思路(简单的生产者消费者模型),共享资源:任务队列,使用run向任务队列中添加任务,线程从任务队列中取任务,执行。

class ThreadPool : noncopyable
{
 public:
  typedef std::function<void ()> Task;

  explicit ThreadPool();
  ~ThreadPool();

  // Must be called before start().
  void setMaxQueueSize(int maxSize) { maxQueueSize_ = maxSize; }
  void setThreadInitCallback(const Task& cb)
  { threadInitCallback_ = cb; }

  void start(int numThreads);
  void stop();

  size_t queueSize() const;
  
  void run(Task f);

 private:
  bool isFull() const REQUIRES(mutex_);
  void runInThread();
  Task take();

  mutable MutexLock mutex_;
  Condition notEmpty_ GUARDED_BY(mutex_);
  Condition notFull_ GUARDED_BY(mutex_);
  Task threadInitCallback_;
  std::vector<std::unique_ptr<muduo::Thread>> threads_;
  std::deque<Task> queue_ GUARDED_BY(mutex_);
  size_t maxQueueSize_;
  bool running_;
};

简单分析下类

  • setMaxQueueSize() : 设置任务队列的大小
  • setThreadInitCallback() : 设置初始化完毕回调,在线程正式运行起来后会调用该回调
  • start() : 启动线程池
  • run(Task f) : 添加任务
    条件变量 notEmpty_ ,notFull_ 最后讲解
源文件

start

void ThreadPool::start(int numThreads)
{
  assert(threads_.empty());
  running_ = true;
  threads_.reserve(numThreads);
  for (int i = 0; i < numThreads; ++i)
  {
    threads_.emplace_back(new muduo::Thread(
        std::bind(&ThreadPool::runInThread, this), name_ + id)); //线程运行的时候回调runInThread
    threads_[i]->start();
  }
  //线程池没有额外设置线程
  if (numThreads == 0 && threadInitCallback_)
  {
    threadInitCallback_();
  }
}
分析

根据numThreads创建Thread对象并运行。runInThread为消费者,从任务队列中取任务运行

代码规范

threads_是vector对象,vector每次扩张,会以当前元素*2扩张,使用reserve,直接>指定了长度,避免了可能的无效内存的分配,。在使用vector时如果,确定vector不会增长,更好的做法是如上述代码一样,直接预分配。

线程消费者任务

// 作为每个线程的任务
void ThreadPool::runInThread()
{
  try
  {
    if (threadInitCallback_)
    {
      threadInitCallback_();//执行线程运行起来的回调函数
    } 
    while (running_)
    {
      Task task(take());
      if (task)
      {
        task();
      }
    }
  }
  catch (...)
  {

  }
}

// 从任务队列获取一个任务
// 如果任务队列为空take将会堵塞,等待直到到任务队列不为空
ThreadPool::Task ThreadPool::take()
{
  MutexLockGuard lock(mutex_);
  // always use a while-loop, due to spurious wakeup
  while (queue_.empty() && running_)
  {
    notEmpty_.wait();//如果任务队列空,挂起该线程
  }
  Task task;
  if (!queue_.empty())
  {
    task = queue_.front();
    queue_.pop_front();
    if (maxQueueSize_ > 0)
    {
      notFull_.notify();
    }
  }
  return task;
}

生产者

//添加任务的,如果queue_满了,则堵塞等到queue被消费后继续添加
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();
  }
}
  while (queue_.empty() && running_)
  {
    notEmpty_.wait();//如果任务队列空,挂起该线程
  }

分析下上面这块代码,while的目的是将该函数变为阻塞式函数,notEmpty_.wait(),挂起该线程,是为了避免对cpu的消耗。当线程([生产者线程)调用notEmpty_.notify(),唤醒调用notEmpty_.wait()堵塞的线程(消费者线程),消费者线程执行while,这时候任务队列不为空了,解除堵塞,执行下面逻辑。

标签:muduo,Thread,void,_.,base,线程,tid,latch
来源: https://blog.csdn.net/qq_38166063/article/details/122483208