LoopThread实例化EventLoop管理类实现(模块八)

本文主要是介绍LoopThread实例化EventLoop管理类实现(模块八)，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

类功能

类定义

类实现

编译测试

类功能

类定义

class LoopThread
{
private:/* 用于实现_loop获取的同步关系,避免线程创建了,但是_loop还没有实例化之前去获取_loop*/std::mutex _mutex;   // 互斥锁std::cond_variable _cond; // 条件变量EventLoop *_loop;    // EventLoop指针变量,这个对象需要在线程内实例化std::thread _thread; // EventLoop对应的线程
private:/*实例化EventLoop对象,唤醒_cond上有可能阻塞的线程,并且开始运行EventLoop模块的功能*/void ThreadEntry();public:/*创建线程,设定线程入口函数*/LoopThread();/*返回当前线程关联的EventLoop对象指针*/EventLoop *GetLoop();
};

类实现

// EventLoop实例化管理类
class LoopThread
{
private:/* 用于实现_loop获取的同步关系,避免线程创建了,但是_loop还没有实例化之前去获取_loop*/std::mutex _mutex;             // 互斥锁std::condition_variable _cond; // 条件变量EventLoop *_loop;              // EventLoop指针变量,这个对象需要在线程内实例化std::thread _thread;           // EventLoop对应的线程
private:/*实例化EventLoop对象,唤醒_cond上有可能阻塞的线程,并且开始运行EventLoop模块的功能*/void ThreadEntry(){EventLoop loop;{std::unique_lock<std::mutex> lock(_mutex);_loop = &loop;_cond.notify_all();}loop.Start();}public:/*创建线程,设定线程入口函数*/LoopThread() : _loop(NULL), _thread(std::thread(&LoopThread::ThreadEntry, this)) {}/*返回当前线程关联的EventLoop对象指针*/EventLoop *GetLoop(){EventLoop *loop = NULL;{std::unique_lock<std::mutex> lock(_mutex); // 加锁_cond.wait(lock, [&](){ return _loop != NULL; }); // loop为NULL就一直阻塞loop = _loop;}return loop;}
};

编译测试

tcp_srv.cc

对tcp_srv.cc测试代码稍加了添加修改

主体的代码参考文章

Acceptor监听套接字管理类实现(模块七)-CSDN博客

#include "../source/server.hpp"EventLoop loop;
// 管理所有的连接
std::unordered_map<uint64_t, PtrConnection> _conns;
uint64_t conn_id = 0;
EventLoop base_loop;    // 主线程
std::vector<LoopThread> threads(2);
int next_loop = 0;void ConnectionDestroy(const PtrConnection &conn)
{_conns.erase(conn->Id());
}
void OnConnected(const PtrConnection &conn)
{DBG_LOG("NEW CONNECTION:%p", conn.get());
}
void OnMessage(const PtrConnection &conn, Buffer *buf)
{DBG_LOG("%s", buf->ReadPosition());buf->MoveReadOffset(buf->ReadAbleSize());std::string str = "Hello World";conn->Send(str.c_str(), str.size());//conn->Shutdown(); // 调用关闭接口
}void NewConnection(int fd)
{conn_id++;next_loop = (next_loop + 1) % 2;PtrConnection conn(new Connection(threads[next_loop].GetLoop(), conn_id, fd));conn->SetMessageCallback(std::bind(OnMessage, std::placeholders::_1, std::placeholders::_2));conn->SetSrvClosedCallback(std::bind(ConnectionDestroy, std::placeholders::_1));conn->SetConnectedCallback(std::bind(OnConnected, std::placeholders::_1));conn->EnableInactiveRelease(10); // 启动非活跃超时销毁conn->Established();             // 就绪初始化_conns.insert(std::make_pair(conn_id, conn));DBG_LOG("NEW ---------------");
}int main()
{srand(time(NULL));Acceptor acceptor(&base_loop, 8500);acceptor.SetAcceptCallback(std::bind(NewConnection, std::placeholders::_1));acceptor.Listen();while (1){base_loop.Start();}return 0;
}

日志添加