文章目录
- Thread
- EventLoopThread
- EventLoopThreadPool
Thread
代码语言：javascript
复制#pragma once
#include "nocopyable.hpp"
#include <functional>
#include <thread>
#include <memory>
#include <unistd.h>
#include <string>
#include <atomic>
class Thread:public nocpoyable{
public:
using ThreadFunc = std::function<void()>;
explicit Thread(ThreadFunc func,const std::string &name);
~Thread();
void start();
void join();
bool started() const{return started_;}
pid_t tid() const{return tid_;}
const std::string& name(){return name_;}
static int numCreateThread(){return numCreateThread_;}
private:
bool started_;
bool joined_;
std::shared_ptr<std::thread> thread_;
pid_t tid_;
ThreadFunc func_;
std::string name_;
static std::atomic_int numCreateThread_;
void setDefaultName();
};
代码语言：javascript
复制#include "thread.hpp"
#include "currenthread.hpp"
#include <semaphore.h>
Thread::Thread(ThreadFunc func,const std::string &name)
:started_(false),
joined_(false),
tid_(0),
func_(std::move(func)),
name_(name)
{
setDefaultName();
}
Thread::~Thread(){
if(started_ && !joined_){
thread_->detach(); //thread有带设置分离方法
}
}
//一个thread对象记录一个线程的详细信息
void Thread::start(){
started_ = true;
sem_t sem;
sem_init(&sem,false,0);
//开启线程
thread_ = std::shared_ptr<std::thread>(new std::thread([&](){
tid_ = CurrentThread::tid(); //获取线程tid
sem_post(&sem);
func_(); //线程任务
}));
//这里等待新线程id
sem_wait(&sem); //线程执行过快，防止在线程创建完成之前函数就返回了
}
void Thread::join(){
joined_ = true;
thread_->join();
}
void Thread::setDefaultName(){
int num = numCreateThread_;
if(name_.empty()){
char buf[32] = {0};
snprintf(buf,sizeof buf,"Thread%d",num);
name_ = buf;
}
}
EventLoopThread
代码语言：javascript
复制#pragma once
#include "nocopyable.hpp"
#include "thread.hpp"
#include<functional>
#include<mutex>
#include<condition_variable>
class EventLoop;
class EventLoopThread : public nocpoyable{
public:
using ThreadInitCallback = std::function<void(EventLoop*)>;
EventLoopThread(const ThreadInitCallback &cb,const std::string &name);
~EventLoopThread();
EventLoop* startloop();
private:
void threadFunc();
EventLoop* loop_;
bool exciting_;
Thread thread_;
std::mutex mutex_;
std::condition_variable cond_;
ThreadInitCallback callback_;
};
代码语言：javascript
复制#include "eventloopthread.hpp"
#include "eventloop.hpp"
EventLoopThread::EventLoopThread(const ThreadInitCallback &cb, const std::string &name)
:loop_(nullptr),exciting_(false),thread_(std::bind(&EventLoopThread::threadFunc,this),name),
mutex_(),cond_(),callback_(cb)
{
}
EventLoopThread::~EventLoopThread(){
exciting_ = true;
if(loop_ != nullptr){
loop_->quit();
thread_.join();
}
}
//每次都是在新线程里执行的
EventLoop* EventLoopThread::startloop(){
thread_.start(); //启动底层新线程
EventLoop* loop = nullptr;
{
std::unique_lock<std::mutex> lock(mutex_);
while(loop == nullptr){
cond_.wait(lock);
}
loop = loop_;
}
return loop;
}
void EventLoopThread::threadFunc(){
EventLoop loop; //创建一个独立的EventLoop
if(callback_){
callback_(&loop);
}
{
std::unique_lock<std::mutex> lock(mutex_);
loop_ = &loop;
cond_.notify_one();
}
loop.loop(); // poller.poll
std::unique_lock<std::mutex> lock(mutex_);
loop_ = nullptr;
}
EventLoopThreadPool
代码语言：javascript
复制#pragma once
#include "nocopyable.hpp"
#include <functional>
#include <string>
#include <vector>
#include <memory>
class EventLoop;
class EventLoopThread;
class EventLoopThreadPool:public nocpoyable{
public:
using ThreadInitCallback = std::function<void(EventLoop*)>;
EventLoopThreadPool(EventLoop *baseloop,const std::string &nameArg);
~EventLoopThreadPool();
void setThreadNum(int numThread){
numCreateThread_ = numThread;
}
void start(const ThreadInitCallback &cb = ThreadInitCallback());
//如果工作在多线程中，baseloop会默认以轮询的方式分配channel给subloop
EventLoop* GetNextLoop();
std::vector<EventLoop*> getAllLoops();
bool started() const{return started_;}
const std::string name(){return name_;}
private:
EventLoop* baseloop_;
std::string name_;
bool started_;
int numCreateThread_;
int next_;
std::vector<std::unique_ptr<EventLoopThread>> thread_;
std::vector<EventLoop*> loops_;
};
代码语言：javascript
复制#include "eventloopthreadpool.hpp"
#include "eventloopthread.hpp"
EventLoopThreadPool::EventLoopThreadPool(EventLoop *baseloop, const std::string &nameArg)
:baseloop_(baseloop),name_(nameArg),started_(false),numCreateThread_(0),next_(0)
{
}
EventLoopThreadPool::~EventLoopThreadPool()
{
}
void EventLoopThreadPool::start(const ThreadInitCallback &cb = ThreadInitCallback()){
started_ = true;
for(int i = 0;i<numCreateThread_; i){
char buf[name_.size() 32];
snprintf(buf,sizeof buf,"%s%d",name_.c_str(),i);
EventLoopThread* t = new EventLoopThread(cb,buf);
thread_.push_back(std::unique_ptr<EventLoopThread>(t));
loops_.push_back(t->startloop()); //创建线程，绑定一个新的EventLoop，并返回该loop地址
}
if(numCreateThread_ == 0 && cb){ //只有一条默认线程
cb(baseloop_);
}
}
//如果工作在多线程中，baseloop会默认以轮询的方式分配channel给subloop
EventLoop *EventLoopThreadPool::GetNextLoop(){
EventLoop* loop = baseloop_;
if(!loops_.empty()){
loop = loops_[next_];
next_;
if(next_ >= loops_.size()){
next_ = 0;
}
}
return loop;
}
std::vector<EventLoop *> EventLoopThreadPool::getAllLoops(){
if(loops_.empty()){
return std::vector<EventLoop*>(1,baseloop_);
}
else{
return loops_;
}
}
