C++-集群聊天室(2):muduo网络库

用 muduo 写一个 TCP 服务器

基于epoll + 线程池

接收客户端连接

客户端发什么，服务器原样返回（Echo）

打印连接、断开、收发数据日志

muduo 的核心思想：

网络 I/O 与业务逻辑彻底解耦

muduo 的核心模型

muduo =Reactor + One Loop Per Thread

整体结构分析

main()
└── EventLoop loop // 主 Reactor（epoll）
└── ChatServer server
└── TcpServer _server
├── ConnectionCallback
├── MessageCallback
└── 线程池 (setThreadNum)

典型 muduo Reactor 模型：

主线程(EventLoop)
|
|--- 接收新连接
|--- 分发连接到子线程(EventLoop)
|
|--- 处理读写事件

整体代码：

/* muduo网络库给用户提供两个主要的类 TcpServer:用户编写服务器程序的 TcpClient:用户编写客户端程序的 epoll + 线程池 好处：能够把网络I/O的代码和业务代码区分开来 用户的连接与断开 用户的可读写事件 */ #include<muduo/net/TcpServer.h> #include<muduo/net/EventLoop.h> #include<iostream> #include<functional> using namespace std; using namespace muduo; using namespace muduo::net; using namespace placeholders; /*基于muduo网络库开发服务器程序 1.组合TcpSever对象 2.创建EventLoop事件循环对象的指针 3.明确TcpServer构造函数需要什么参数，输出ChatServer的构造函数 */ class ChatServer { public: ChatServer(EventLoop * loop,const InetAddress &listenAddr,const string &nameArg):_server(loop,listenAddr,nameArg),_loop(loop) { _server.setConnectionCallback(bind(&ChatServer::onConnection,this,_1)); _server.setMessageCallback(bind(&ChatServer::onMessage,this,_1,_2,_3)); //设置EventLoop的线程个数 _server.setThreadNum(10); } //启动ChatServer服务 void start() { _server.start(); } private: //TcpServer绑定的回调函数，当有新连接或连接中断时调用 void onConnection(const TcpConnectionPtr &conn) { if(conn->connected()) { cout<<conn->peerAddress().toIpPort()<< " -> " << conn->localAddress().toIpPort()<<"state:online"<<endl; } else{ cout<<conn->peerAddress().toIpPort()<< " -> " << conn->localAddress().toIpPort()<<"state:offline"<<endl; conn->shutdown();//close(fd) _loop->quit(); } } //TcpServer绑定的回调函数，当有新数据时调用 void onMessage(const TcpConnectionPtr& conn, Buffer* buffer, Timestamp time) { string buf=buffer->retrieveAllAsString(); cout<<"recv data:"<<buf<<"time:"<<time.toString()<<endl; conn->send(buf); } TcpServer _server; EventLoop* _loop; }; int main() { EventLoop loop;//epoll InetAddress addr("127.0.0.1",6000); ChatServer server(&loop,addr,"ChatServer"); server.start();//listen epoll_ctl=>epoll loop.loop();//epoll_wait以阻塞方式等待新用户连接，已连接用户的读写事件等 return 0; }

头文件 & 命名空间

#include <muduo/net/TcpServer.h> #include <muduo/net/EventLoop.h> #include <functional> using namespace muduo; using namespace muduo::net; using namespace placeholders;

EventLoop（Reactor 的本体）

EventLoop loop;

EventLoop 是什么？

EventLoop = epoll + 事件分发器

loop.loop() 在干什么？

void EventLoop::loop() { while (!quit_) { activeChannels_ = poller_->poll(); // epoll_wait for (Channel* ch : activeChannels_) { ch->handleEvent(); } } }

为什么 EventLoop 不能拷贝？

内部持有 fd

绑定线程 ID

一个 loop 只能在一个线程里跑

TcpServer

TcpServer 本质是什么？

TcpServer = Acceptor + 线程池 + 连接管理器

class TcpServer { EventLoop* loop_; // 主 loop Acceptor acceptor_; // 监听 socket EventLoopThreadPool threadPool_; map<string, TcpConnectionPtr> connections_; };

TcpServer 的职责

InetAddress（地址封装）

InetAddress addr("127.0.0.1", 6000);

等价于：

sockaddr_in addr; addr.sin_family = AF_INET; addr.sin_port = htons(6000); addr.sin_addr.s_addr = inet_addr("127.0.0.1");

ChatServer 构造函数

ChatServer(EventLoop* loop, const InetAddress& listenAddr, const string& nameArg) : _server(loop, listenAddr, nameArg), _loop(loop)

创建 TcpServer
内部创建监听 socket
但 还没 listen

回调机制

_server.setConnectionCallback( bind(&ChatServer::onConnection, this, _1) );

muduo 什么时候调用它？

✔ accept 成功
✔ TCP 连接建立
✔ TCP 连接断开

TcpConnectionPtr

const TcpConnectionPtr& conn

等价于

shared_ptr<TcpConnection>

为什么一定要 shared_ptr？

连接可能：

正在读

正在写

正在关闭

防止提前析构

保证回调执行期间对象存在

这是 muduo 稳定性的核心设计

setMessageCallback

_server.setMessageCallback( bind(&ChatServer::onMessage, this, _1, _2, _3) );

什么时候触发？

socket fd EPOLLIN

onMessage 参数拆解

const TcpConnectionPtr& conn Buffer* buffer Timestamp time

Buffer 是什么？

muduo 的用户态缓冲区

socket → kernel buffer → Buffer → 用户
Timestamp

time.toString()

= 消息到达服务器的时间
= muduo 在 epoll 返回时打的时间戳

send()

conn->send(buf);

send 是怎么做到非阻塞的？

尝试 write
写不完 → 放入 output buffer
注册 EPOLLOUT
可写时继续写

main 函数执行全过程

main
EventLoop loop
ChatServer server
server.start()
├ socket()
├ bind()
├ listen()
├ 创建线程池
loop.loop()
└ epoll_wait (阻塞)