仿muduo库实现并发服务器（3）
LoopThread模块
目的
将EventLoop模块与线程整合一起，让EventLoop模块与线程是一一对应的，这里就延伸出一个问题：是要先在主线程创建EventLoop对象，再分配给子线程还是先创建子线程，再在子线程里创建EventLoop对象？
含义
在回答上面的疑问前要先清楚：在eventloop里进行一个操作时，往往会涉及判断当前操作是否在eventloop模块对应的线程中运行（将当前线程ID和eventloop的成员_thread_id作比较，相同就表示在同一线程可以被理解处理，不同就表示当前线程并不是与eventloop对象绑定的那个线程，就需要把这个操作压入eventloop对象任务队列）；
Eventloop实例化对象，在构造时就会初始化_thread_id，这个id就是当前线程id。
先创建eventloop对象再创建子线程，将子线程的线程id再赋值给eventloop的_thread_id，这个期间将是不可控的：很可能出现
（1）创建eventloop对象时_thread_id为主线程id，并且调用了_loop->RunInLoop(task)，此时当前线程id等于_thread_id，任务task直接执行；
（2）接着创建子线程将子线程id赋值给了_thread_id，子线程开始运行读取Channel列表查看触发什么事件，而刚刚的任务要是没有跑完并且尝试去修改Channel列表，就变成了两个线程同时对同一块进行操作，一个进行“写操作”一个进行“读操作”，程序会直接崩溃。
所以LoopThread模块就要求，eventloop在实例化对象时必须在线程内部，就是要先创建线程，然后在线程的入口函数去实例化eventloop对象，不要出现对_thread_id重新赋值的情况。
设计思想
- 创建线程
- 在线程中实例化eventloop 对象
- 可以向外部返回实例化后的eventloop：给connection初始化作为参数使用
代码设计
// 让eventloop与线程一一对应
class EventLoopThread
{
private:
std::mutex _mutex; // 互斥锁
std::condition_variable _cond; //条件变量
EventLoop *_loop; // EventLoop需在线程内部实例化
std::thread _thread; // EventLoop 对应的线程
private:
//实例化对象，唤醒_cond上有可能阻塞的线程，运行loop模块的功能
void ThreadEntry()
{
EventLoop loop;
{
std::unique_lock _lock(_mutex);//保护_loop
_loop = &loop;
_cond.notify_all();
}
loop.Start();
}
public:
//创建线程，执行线程入口函数ThreadEntry()
EventLoopThread() : _loop(nullptr), _thread(&EventLoopThread::ThreadEntry, this){}
//返回当前线程关联的eventloop指针
EventLoop *GetLoop()
{
if (_loop == nullptr)
{
//避免_loop还没初始化就被获取
std::unique_lock _lock(_mutex); //wait需要lock配套使用
_cond.wait(_lock, [&](){ return _loop != nullptr; });//_loop为空就一直阻塞
}
return _loop;
}
};
下面对tcp_srv.cpp进修改，主要是给不同的线程分配不同的任务：一个负责监听部分一个负责通信部分，实现效率提升
#include "../source/server.hpp"
uint64_t conn_id = 0;
std::unordered_map _conns;
EventLoop loop;
std::vector _loop_threads(2);
int next_loop = 0;
void ConnectionDestroy(const PtrConnection& ptr)
{
_conns.erase(ptr->Id());
}
void OnConnected(const PtrConnection& ptr)
{
DBG_LOG("New Connection: %p", ptr.get());
}
void OnMessage(const PtrConnection& ptr, Buffer* buf)
{
DBG_LOG("%s", buf->ReadPosition());
buf->MoveReadOffset(buf->ReadAbleSize());
std::string s = "hello, bugubugu!";
ptr->Send(s.c_str(), s.size());
ptr->Shutdown();
}
void HandleNewConnection(int newfd)
{
conn_id++;
next_loop = (next_loop + 1) % 2;
PtrConnection ptr_con(new Connection(conn_id, newfd, _loop_threads[next_loop].GetLoop()));
_conns.insert(std::make_pair(conn_id, ptr_con));
ptr_con->SetConnectedCallback(std::bind(OnConnected, std::placeholders::_1));//参数由connection内部提供
ptr_con->SetMessageCallback(std::bind(OnMessage, std::placeholders::_1, std::placeholders::_2));
ptr_con->SetServerClosedCallback(std::bind(ConnectionDestroy, std::placeholders::_1));
//10s后对非活跃连接进行释放
//释放任务的添加要放在监控之前，如果先开启监控，立即有事件，在刷新释放任务里找不到这个任务
ptr_con->EnableInactiveRelease(10);
ptr_con->Established();
DBG_LOG("是否主线程运行？？？？");
}
int main()
{
Acceptor acceptor(&loop, 8080);
acceptor.SetAcceptCallback(std::bind(HandleNewConnection, std::placeholders::_1));
acceptor.Listen();
while(1)
{
loop.Start();
}
return 0;
}
主从reactor的作用：主线程的eventloop负责监听新连接，以及调用对新连接的处理函数
void HandleNewConnection(int newfd) ，将其分发给从属线程：Connection 对象在构造时绑定的 _loop 指针指向的是通过 threads[next_loop].GetLoop（） 获取的从属线程 EventLoop。这一步就将新连接connection和从属线程 EventLoop绑定在一起。
后面从属线程 EventLoop就只负责通信的监听事件、对触发事件的处理以及业务处理，不会出现一个线程一直在处理业务而无法及时处理新连接的产生，效率提高。
LoopThreadPool模块
目的
对所有的LoopThread 进行管理和分配
功能
1、线程数量可配置（0个或多个）：上面已经说明了主从Reactor 模型：主线程只负责新连接获取，从属线程负责新连接的事件监控以及处理。也存在一种情况：从属线程数量为0，那么主线程既负责新连接获取、也负责新连接的事件监控以及处理，相当于实现单Reactor服务器
2、对所有线程进行管理：管理0个或者多个LoopThread对象
3、提供线程分配的功能：当主线程获取新连接，需要将连接挂到从属线程上进行事件监控以及处理，就是将线程一一对应的eventloop返回，供Connection初始化。这里存在两种情况：
（1）从属线程数量为0，将连接直接分配给主线程的EventLoop
（2）从属线程数量有多个，采用RR轮转思想，进行线程分配。
代码设计
// 管理EventLoopThread
class LoopThreadPool
{
private:
int _thread_count; // 从属线程的数量
int _next_loop;
EventLoop *_baseloop; // 在主线程运行
std::vector _threads; // 保存所有的从属线程信息
std::vector _loops; // 从属线程数量大于0才进行eventloop的分配
public:
LoopThreadPool(EventLoop *baseloop) : _baseloop(baseloop), _next_loop(0), _thread_count(0) {}
// 线程数量可配置
void SetThreadCount(int count) { _thread_count = count; }
// 创建所有的从属线程
void CreateChlids()
{
if (_thread_count > 0)
{
_threads.resize(_thread_count);
_loops.resize(_thread_count);
for (int i = 0; i < _thread_count; i++)
{
_threads[i] = new EventLoopThread;
_loops[i] = _threads[i]->GetLoop();
}
}
return;
}
// 进行eventloop的分配
EventLoop *NextLoop()
{
if (_thread_count == 0)
return _baseloop;
_next_loop = (_next_loop + 1) % _thread_count;
return _loops[_next_loop];
}
};
TcpServer模块
含义
是前面所有模块的整合，通过TcpServer类实例化的对象直接完成一个服务器的搭建
下面按照前面的tcp_srv.cc的思路流程，不难对TcpServer类进行声明
管理
1、通过Accpetor对象，创建一个监听套接字
2、创建一个EventLoop对象-->baseloop变量，实现对监听套接字的事件监控及处理
3、std::unordered_map
4、创建一个LoopThreadPool对象-->loop_pool线程池，通过分配eventloop对新建连接进行事件监控及处理
功能
1、设置从属线程数量
2、设置各种回调函数（包括连接建立完成、收到消息、关闭连接、任意事件）：用户/组件使用者先设置给TcpServer，TcpServer设置给获取的新连接（Connection）
3、是否启动非活跃连接超时销毁功能
4、添加定时任务（可以不使用，有用户决定；这个定时任务是在baseloop对应线程执行，而非活跃连接超时销毁任务一般是在从属loop对应线程执行）
5、启动服务器
注意上面的功能是提供给用户使用，而还有一些功能是被包含在上面的功能中的，比如对获取的新连接处理方法、添加/移除新连接的管理信息等等。
代码设计
class TcpServer
{
private:
EventLoop _baseloop; // 主线程的EventLoop，负责监听连接以及处理
int _port;
Acceptor _acceptor; // 监听套接字的管理对象
bool _enable_inactive_release; // 是否启动非活跃连接超时销毁的判断标志
int _timeout; // 非活跃连接的统计时间，即多长时间无通信就是非活跃连接
LoopThreadPool _pool; // 从属线程池
uint64_t _next_id; // 自动增长id，用于定时任务id、connection对象(connection对象里的超时销毁任务id)
std::unordered_map _conns;
using ConnectedCallback = std::function;
using MessageCallback = std::function;
using ClosedCallback = std::function;
using AnyEventCallback = std::function;
ConnectedCallback _connected_callback;
MessageCallback _message_callback;
ClosedCallback _closed_callback;
AnyEventCallback _anyevent_callback;
using Functor = std::function;
private:
void RemoveConnectionInLoop(const PtrConnection &ptr)
{
auto it = _conns.find(ptr->Id());
if(it != _conns.end()) _conns.erase(it);
}
//从管理connection的_conns中移除连接信息
void RemoveConnection(const PtrConnection &ptr)
{
_baseloop.RunInLoop(std::bind(&TcpServer::RemoveConnectionInLoop, this, ptr));
}
//为新连接构造一个connection进行管理
void HandleNewConnection(int newfd)
{
_next_id++;
PtrConnection ptr_con(new Connection(_next_id, newfd, _pool.NextLoop()));
_conns.insert(std::make_pair(_next_id, ptr_con));
ptr_con->SetConnectedCallback(_connected_callback); // 参数由connection内部提供
ptr_con->SetMessageCallback(_message_callback);
ptr_con->SetAnyEventCallback(_anyevent_callback);
ptr_con->SetClosedCallback(_closed_callback);
ptr_con->SetServerClosedCallback(std::bind(&TcpServer::RemoveConnection, this, std::placeholders::_1));
// _timeout s后对非活跃连接进行释放
// 释放任务的添加要放在监控之前，如果先开启监控，立即有事件，在刷新释放任务里找不到这个任务
if(_enable_inactive_release == true) ptr_con->EnableInactiveRelease(_timeout);
ptr_con->Established();
DBG_LOG("是否主线程运行？？？？");
}
void RunAfterInLoop(const Functor & task, int delay)
{
_next_id++;
_baseloop.TimerAdd(_next_id, delay, task);
}
public:
TcpServer(int port) : _port(port),
_acceptor(&_baseloop, port),
_enable_inactive_release(false),
_next_id(0),
_pool(&_baseloop)
{
_acceptor.SetAcceptCallback(std::bind(&TcpServer::HandleNewConnection, this,std::placeholders::_1));
_acceptor.Listen();
}
void SetThreadCount(int count)
{
_pool.SetThreadCount(count);
}
void SetConnectedCallback(const ConnectedCallback &cb) { _connected_callback = cb; }
void SetMessageCallback(const MessageCallback &cb) { _message_callback = cb; }
void SetClosedCallback(const ClosedCallback &cb) { _closed_callback = cb; }
void SetAnyEventCallback(const AnyEventCallback &cb) { _anyevent_callback = cb; }
void EnableInactiveRelease(int timeout)
{
_enable_inactive_release = true;
_timeout = timeout;
}
//在baseloop添加定时任务
void RunAfter(const Functor & task, int delay)
{
_baseloop.RunInLoop(std::bind(&TcpServer::RunAfterInLoop, this, task, delay));
}
void Start()
{
_pool.CreateChlids();
_baseloop.Start();
}
};
实现流程
- 在TcpServer中实例化一个Acceptor对象、一个EventLoop对象（baseloop）并给Acceptor设置可读回调函数
- 将Acceptor 挂到baseloop上进行事件监控
- 一旦Acceptor对象就绪可读事件，执行可读事件回调函数获取新连接
- 对新连接创建一个Connection进行管理，设置功能回调函数（连接完成、消息获取、连接关闭、任意事件）
- 启动Connection的非活跃连接超时销毁任务
- 将新连接对应的Connection挂到LoopThreadPool中的从属线程对应的EventLoop进行事件监控
- 一旦Connection对象就绪可读事件，执行可读事件回调函数，读取数据，读取数据后调用TcpServer设置的消息回调
测试代码修改（tcp_svr.cc）
#include "../source/server.hpp"
void OnConnected(const PtrConnection& ptr)
{
DBG_LOG("New Connection: %p", ptr.get());
}
void OnMessage(const PtrConnection& ptr, Buffer* buf)
{
DBG_LOG("%s", buf->ReadPosition());
buf->MoveReadOffset(buf->ReadAbleSize());
std::string s = "hello, bugubugu!";
ptr->Send(s.c_str(), s.size());
//ptr->Shutdown();
}
void OnClosed(const PtrConnection& ptr)
{
DBG_LOG("Close Connection: %p", ptr.get());
}
int main()
{
TcpServer svr(8080);
svr.SetClosedCallback(OnClosed);
svr.SetConnectedCallback(OnConnected);
svr.SetMessageCallback(OnMessage);
svr.SetThreadCount(2);
svr.EnableInactiveRelease(10);
svr.Start();
return 0;
}
这里的测试结果和前面没有实现TcpServer模块结果几乎一样：唯一不同的是 Release Connection: 0x56083a72f740 执行的线程不一样
上面是移除连接信息没有被RunInLoop的打印界面
下面是移除连接信息被RunInLoop的打印界面
造成执行Connection析构函数的线程不一样的原因是
TcpServer里对于从管理connection的_conns中移除连接信息这个模块，为保证线程安全放在主线程运行
void RemoveConnection(const PtrConnection &ptr)
{
_baseloop.RunInLoop(std::bind(&TcpServer::RemoveConnectionInLoop, this, ptr));
}
谁最后释放 PtrConnection谁就执行Connection的析构函数，这里很明显就是 _conns变量。
（1）之前RemoveConnection是由从属线程执行，当erase执行后，PtrConnection减到一时，析构函数执行；
（2）现在RemoveConnection还是由从属线程执行，但是走到_baseloop.RunInLoop 这一块发现当前线程id不是baseloop的id，那么线程就会把任务加入 _baseloop的任务队列并唤醒_baseloop对应的主线程进行监控，主线程执行任务队列的RemoveConnectionInLoop，那么conns_.erase 从映射表中移除连接，PtrConnection减到一时，析构函数执行。
注意：SIGPIPE:当有一方主动关闭连接时，另一方发送信息会触发SIGPIPE异常，导致发送方程序退出，所以这里告诉操作系统忽略这个信号。当底层的 write 或 send 函数在遭遇对方关闭连接的情况时，将不会导致进程死亡
class NetWork
{
public:
NetWork()
{
DBG_LOG("SIGPIPE INIT");
signal(SIGPIPE, SIG_IGN);
}
};
static NetWork nw;
EchoServer回显服务器
继续对TcpServer模块进行封装，把TcpServer的设置回调函数功能、设置从属线程个数已经启动非活跃销毁功能在EchoServer内部实现
代码设计
class EchoServer
{
private:
TcpServer _server;
private:
void OnConnected(const PtrConnection &ptr)
{
DBG_LOG("New Connection: %p", ptr.get());
}
void OnMessage(const PtrConnection &ptr, Buffer *buf)
{
ptr->Send(buf->ReadPosition(), buf->ReadAbleSize());
buf->MoveReadOffset(buf->ReadAbleSize());
// ptr->Shutdown();
}
void OnClosed(const PtrConnection &ptr)
{
DBG_LOG("Close Connection: %p", ptr.get());
}
public:
EchoServer(int port) : _server(port)
{
_server.SetClosedCallback(std::bind(&EchoServer::OnClosed, this, std::placeholders::_1));
_server.SetConnectedCallback(std::bind(&EchoServer::OnConnected, this, std::placeholders::_1));
_server.SetMessageCallback(std::bind(&EchoServer::OnMessage, this, std::placeholders::_1, std::placeholders::_2));
_server.SetThreadCount(2);
_server.EnableInactiveRelease(10);
}
void Start()
{
_server.Start();
}
};
注意这里设置回调函数一定要用bind，因为OnConnected、OnMessage、OnClosed等函数的实现是类内函数，和之前直接在测试代码里面实现不一样（全局函数），这里需要传this指针。
//main.cc
#include "echo.hpp"
int main()
{
EchoServer _server(8080);
_server.Start();
return 0;
}
再次封装后，只需要传个端口号已经启动Start函数就可以实现一个简单的回显服务器，即把客户端发送的信息再次发送给客户端。
模块关系图
