muduo学习笔记线程类

learn_muduo

线程属性

线程标识 pthreadId_，pid_t
线程函数 func_
线程名字 name_
线程序号 numCreated_

bool             started_;       // 线程状态标识
bool            joined_;
pthread_t       pthreadId_;     // pthread_函数使用
pid_t           tid_;           // 线程标识
ThreadFunc      func_;          // 线程函数
string          name_;          // 线程名字
CountDownLatch  latch_;         // 倒计时
static AtomicInt32 numCreated_; // 线程序号

pthread_t的值很大，无法作为一些容器的key值。 glibc的Pthreads实现实际上把pthread_t作为一个结构体指针，指向一块动态分配的内存，但是这块内存是可以反复使用的，也就是说很容易造成pthread_t的重复。也就是说pthreads只能保证同一进程内，同一时刻的各个线程不同；不能保证同一个进程全程时段每个线程具有不同的id，不能保证线程id的唯一性。

在LINUX系统中，建议使用gettid()系统调用的返回值作为线程id，这么做的原因：
返回值是一个pid_t，其值是一个很小的整数，方便输出。
在linux系统中，它直接标识内核任务调度id，可通过/proc文件系统中找到对应项：/proc/tid 或者 /proc/pid/task/tid，方便定位到具体线程。任何时刻都是唯一的，并且由于linux分配新的pid采用递增轮回办法，短时间内启动多个线程也会具有不同的id。

执行过程

new Thread

通过new Thread(&threadFunc)初始化创建Thread

Thread::Thread(ThreadFunc func, const string& n): started_(false),joined_(false),pthreadId_(0),tid_(0),func_(std::move(func)),name_(n),latch_(1)
{setDefaultName();
}

在Thread的构造函数中，初始化线程属性。同时调用setDefaultName()设置线程的名字name_

void Thread::setDefaultName()
{int num = numCreated_.incrementAndGet();if (name_.empty()) {char buf[32];snprintf(buf, sizeof buf, "Thread%d", num);name_ = buf;}
}

start

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;// LOG_SYSFATAL << "Failed in pthread_create";} else {latch_.wait();assert(tid_ > 0);}
}

start()在堆上创建ThreadData对象，作为线程函数的参数，然后通过pthread_create()创建并启动线程。

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

当pthread_creat()创建成功，接着是倒计时类latch_.wait()，初始的 count_ = 1，主线程阻塞到mutex_上等待线程的结束。

void CountDownLatch::wait()
{MutexLockGuard lock(mutex_);while (count_ > 0) {condition_.wait();}
}

void wait()
{MutexLock::UnassignGuard ug(mutex_);int ret = pthread_cond_wait(&pcond_, mutex_.getPthreadMutex()); // 将线程添加到条件变量assert(ret == 0);
}

runInThread

线程函数的执行是通过ThreadData中的runInThread调用func_()完成的。

void runInThread()
{*tid_ = muduo::CurrentThread::tid();tid_ = NULL;latch_->countDown(); // 倒计时减1latch_ = 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());} catch (...) {muduo::CurrentThread::t_threadName = "crashed";fprintf(stderr, "unknow exception caught in Thread %s\n", name_.c_str());throw;}
}

当子线程开始执行，倒计时的latch_.count_ == 0，就会调用condition.notrifyAll()唤醒阻塞的主线程。

void notifyAll()
{int ret = pthread_cond_broadcast(&pcond_);assert(ret == 0);
}

测试程序

依次使用1到8个线程，每个线程向vector中push1千万个数，测试他们需要的时间

#include "muduo/base/Timestamp.h"
#include "muduo/base/Mutex.h"
#include "muduo/base/Thread.h"
#include "muduo/base/CountDownLatch.h"#include <iostream>
#include <stdio.h>
#include <boost/bind.hpp>
#include <boost/ptr_container/ptr_vector.hpp>
#include <vector>muduo::MutexLock g_mutex; // 声明锁
std::vector<int> g_vec;
const int kCount = 10000000; // 每次插入1000w个数void threadFunc()
{for (int i = 0; i < kCount; ++i) {muduo::MutexLockGuard lock(g_mutex); // 上锁g_vec.push_back(i);}
}void test_Mutex()
{std::cout << "---- Mutex ----\n";const int kMaxThreads = 8; // 最多8个线程g_vec.reserve(kMaxThreads * kCount); // 提前分配大小muduo::Timestamp start(muduo::Timestamp::now()); // 当前时间戳// 单个线程不用锁的时间for (int i = 0; i < kCount; ++i) {g_vec.push_back(i);}printf("1 thread(s) without lock %f\n", muduo::timeDifference(muduo::Timestamp::now(), start));for (int i = 0; i < kMaxThreads; ++i) {// i个线程用锁的时间boost::ptr_vector<muduo::Thread> threads;g_vec.clear();start = muduo::Timestamp::now(); // 更新时间戳for (int j = 0; j <= i; ++j) {threads.push_back(new muduo::Thread(&threadFunc)); // 创建线程threads.back().start(); // 启动线程  }for (int j = 0; j <= i; ++j) {threads[j].join(); // 回收线程}printf("%d thread(s) without lock %f\n", i+1, muduo::timeDifference(muduo::Timestamp::now(), start));}
}int main() {test_Mutex();return 0;
}/*
---- Mutex ----
1 thread(s) without lock 0.390272
1 thread(s) with lock 1.485214
2 thread(s) with lock 10.677879
3 thread(s) with lock 11.748183
4 thread(s) with lock 16.022083
5 thread(s) with lock 19.676071
6 thread(s) with lock 23.740399
7 thread(s) with lock 27.879850
8 thread(s) with lock 32.507374
*/