目录前言一、为什么选择 Reactor1.1 一连接一线程的问题1.2 从 epoll 到 Reactor二、先认识服务端用到的几个对象2.1 EventLoop2.2 TcpServer2.3 TcpConnection2.4 Buffer三、实现词典服务器3.1 DictServer.h3.2 DictServer.cpp四、实现客户端4.1 EventLoopThread4.2 为什么还需要 CountDownLatch4.3 DictClient.h4.4 DictClient.cpp五、编译和运行写在最后前言系列C RPC 框架从设计到实现第一篇项目源码JSON-RPChttps://gitee.com/kuang-zhenting/json-rpc上一篇我们已经把Json::Value和 JSON 字符串之间的转换跑通了。不过数据如果一直停留在当前进程里还谈不上远程调用。接下来要解决的问题就是客户端怎么把数据发送给服务端服务端又怎么把处理结果送回来这一篇先不急着写 Json-Rpc 的消息协议而是使用Muduo搭一个很小的词典服务器和客户端跑通一次完整的 TCP 通信。我们主要关注几个问题服务端怎样同时管理多个连接Muduo 的EventLoop、TcpServer、TcpConnection和Buffer分别负责什么客户端为什么需要EventLoopThreadTcpClient::connect()是异步的业务代码怎样等待连接真正建立数据到达以后回调函数是怎样被触发的。等这条最基础的网络链路跑通以后后面再把普通字符串换成真正的 RPC 消息。一、为什么选择 Reactor1.1 一连接一线程的问题最直观的 TCP 服务器模型是Client 1 → Thread 1 → recv / send Client 2 → Thread 2 → recv / send Client 3 → Thread 3 → recv / send ...这种模型很好理解每来一个连接就创建一个线程专门处理它。问题在于一个 TCP 连接建立以后可能很长时间都没有数据。如果连接一直存在对应线程通常也会一直存在。连接数量增加后线程的栈空间、调度和上下文切换都会变成额外成本。因此网络服务器更常见的思路是用少量线程管理大量连接只在某个连接真正产生 I/O 事件时再处理它。1.2 从 epoll 到 Reactor在 Linux 下可以把很多文件描述符交给epoll统一监控。线程不再阻塞在某一个socket上而是等待哪些 socket 现在可以读哪些 socket 现在可以写有没有新的连接到来Reactor 则是在这个基础上进一步组织程序结构等待事件 ↓ 事件就绪 ↓ 找到对应连接 ↓ 执行对应回调Muduo 就是一套基于事件驱动思想的 C TCP 网络库。它很重要的一条设计思路是one loop per thread也就是一个 I/O 线程围绕一个EventLoop工作。Muduo 还支持主 Reactor 接收连接再把已经建立的连接分配给其他 I/O EventLoop。不过我们这一篇的词典 Demo 不需要一开始就配置很多工作线程重点只是把基础收发流程看明白。后面真正做 Json-Rpc 时我们也不需要重新手写epoll、事件循环和网络缓冲区而是把精力放到消息、协议和 RPC 业务本身。二、先认识服务端用到的几个对象搭建词典服务器时不需要把 Muduo 的所有 API 一次学完。先认识下面四个对象就够了。对象当前阶段主要作用EventLoop驱动事件循环TcpServer监听端口、管理服务端连接并注册回调TcpConnection表示一条已经建立的 TCP 连接Buffer保存网络层已经收到、但业务还没有消费的字节2.1 EventLoop当前最重要的两个接口void loop(); void quit();loop()会持续等待并处理 I/O、定时器等事件。服务器启动后通常会进入_baseloop.loop();只要没有退出事件循环服务器就会继续处理网络事件。2.2 TcpServerTcpServer是服务端的入口,这一篇主要会用void setConnectionCallback(...); void setMessageCallback(...); void start();其中ConnectionCallback连接建立或断开时触发MessageCallback某条连接收到数据时触发。可以把它理解成TcpServer 负责“什么时候有事情发生”业务回调负责“事情发生以后做什么”2.3 TcpConnection连接回调和消息回调都会拿到muduo::net::TcpConnectionPtr它本质上是指向TcpConnection的智能指针。当前最常用的是bool connected() const; void send(...);这里要建立一个很重要的认识TcpServer和TcpClient主要负责管理连接流程真正发送业务数据时使用的是对应的TcpConnection。这个认识后面会直接延续到我们的 Json-Rpc 网络抽象层。2.4 BufferMuduo 检测到 socket 可读以后会把数据放进Buffer再调用消息回调。但是 TCP 是字节流一次回调并不等于“一条完整业务消息”。因此这一篇虽然还没有设计最终的LVProtocol但为了让词典 Demo 稳定我们使用一个非常简单的规则每个单词后面加一个 \n也就是用换行符临时表示一条输入结束。这样服务器就可以通过findEOL()判断一整行是否已经到齐。常见的 Buffer 操作可以先记住下面几个操作含义peek()查看当前读位置不消费数据findEOL()查找换行符retrieveUntil()消费到指定位置retrieveAsString(len)取出指定长度并消费readInt32()读取一个整数并推进读位置peekInt32()查看整数但不推进读位置后面我们解析 RPC 长度字段时会再次用到“先看数据够不够再正式消费”这个思路。三、实现词典服务器我们的 Demo 很简单客户端发送Hello 服务端返回你好​ 客户端发送World 服务端返回世界 ​客户端发送其他内容 服务端返回Unknown。这里故意不引入 JSON也不使用最终的 Json-Rpc 消息。我们只验证Muduo 的监听、连接、Buffer、回调和发送链路能够正常工作。完整的数据流可以先看一下3.1DictServer.h#pragma once #include muduo/base/Timestamp.h #include muduo/net/Buffer.h #include muduo/net/EventLoop.h #include muduo/net/TcpConnection.h #include muduo/net/TcpServer.h #include string class DictServer { public: explicit DictServer(int port); void start(); private: void onConnection( const muduo::net::TcpConnectionPtr conn); void onMessage( const muduo::net::TcpConnectionPtr conn, muduo::net::Buffer *buf, muduo::Timestamp timestamp); std::string translate( const std::string word) const; private: // 声明在 _server 前面保证构造 _server 时 _baseloop 已经存在 muduo::net::EventLoop _baseloop; muduo::net::TcpServer _server; };这个类的职责很清楚构造函数创建TcpServer并注册回调onConnection处理连接建立和关闭onMessage读取请求并发送响应translate只负责词典查询不关心网络。3.2DictServer.cpp#include DictServer.h #include muduo/net/InetAddress.h #include functional #include iostream #include unordered_map DictServer::DictServer(int port) : _server( _baseloop, muduo::net::InetAddress(port), DictServer, muduo::net::TcpServer::kReusePort) { _server.setConnectionCallback( std::bind( DictServer::onConnection, this, std::placeholders::_1)); _server.setMessageCallback( std::bind( DictServer::onMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)); } void DictServer::onConnection( const muduo::net::TcpConnectionPtr conn) { std::cout (conn-connected() ? CONNECTION UP : CONNECTION DOWN) std::endl; } std::string DictServer::translate( const std::string word) const { static const std::unordered_map std::string, std::string dict { {Hello, 你好}, {你好, Hello}, {World, 世界}, {世界, World}, }; const auto it dict.find(word); if (it dict.end()) { return Unknown; } return it-second; } void DictServer::onMessage( const muduo::net::TcpConnectionPtr conn, muduo::net::Buffer *buf, muduo::Timestamp timestamp) { (void)timestamp; // 一次回调里可能有多行完整数据所以持续处理 while (true) { const char *eol buf-findEOL(); if (eol nullptr) { // 还没有收到完整一行保留数据等待下一次网络事件 break; } // eol 指向 \n这里先取出换行符之前的内容 std::string word( buf-peek(), eol); // 连同 \n 一起消费 buf-retrieveUntil(eol 1); // 如果客户端发送的是 \r\n顺手去掉末尾的 \r if (!word.empty() word.back() \r) { word.pop_back(); } std::string response translate(word); // 响应同样用 \n 表示一条消息结束 response.push_back(\n); conn-send(response); } } void DictServer::start() { _server.start(); _baseloop.loop(); } int main() { DictServer server(8080); server.start(); return 0; }这里最值得注意的是onMessage()。我们没有使用buf-retrieveAllAsString();然后默认“这次收到的全部数据刚好就是一个单词”。因为 TCP 并不保证这一点。客户端连续发送Hello\n World\n服务端可能一次收到两条也可能先收到半条。因此我们让不完整的数据继续留在 Buffer 中只有找到\n以后才消费一条完整输入。这还是一个非常简单的协议但它已经帮我们建立了后面处理半包和粘包时需要的正确思路。四、实现客户端服务端的网络循环直接运行在主线程没有问题因为服务器主线程本来就要一直处理网络事件。客户端不一样。我们的客户端还希望主线程不断执行std::getline(std::cin, msg);如果主线程直接进入loop.loop();就没办法继续读取用户输入了。因此需要把客户端网络事件循环放到另一个线程中。4.1 EventLoopThreadMuduo 已经提供muduo::net::EventLoopThread通过_baseloop _loopthread.startLoop();它会启动后台线程并返回这个线程内部的EventLoop*。于是客户端形成两个执行方向主线程读取用户输入、发起业务操作​ 后台线程EventLoop 处理网络事件和回调4.2 为什么还需要 CountDownLatch这里还有一个很容易踩坑的地方TcpClient::connect()是异步发起连接。也就是说调用_client.connect();以后这个函数可以先返回但 TCP 连接可能还没有真正建立_conn也还没有在连接回调中保存下来。如果业务代码马上发送connect() 返回 ↓ 连接回调还没有执行 ↓ _conn 仍然为空 ↓ 无法发送因此当前 Demo 使用muduo::CountDownLatch让connect()等到连接回调真正执行以后再返回。它解决的不是“把 Muduo 改成同步网络库”而是在当前调用流程中明确等待“连接已经建立”这个异步事件完成。这个模式在我们当前项目源码里的MuduoClient中也会继续使用。4.3DictClient.h#pragma once #include muduo/base/CountDownLatch.h #include muduo/base/Timestamp.h #include muduo/net/Buffer.h #include muduo/net/EventLoopThread.h #include muduo/net/TcpClient.h #include muduo/net/TcpConnection.h #include string class DictClient { public: DictClient( const std::string ip, int port, bool wait_connect true); void connect(); void shutdown(); void translate( const std::string word); private: void onConnection( const muduo::net::TcpConnectionPtr conn); void onMessage( const muduo::net::TcpConnectionPtr conn, muduo::net::Buffer *buf, muduo::Timestamp timestamp); private: bool _wait_connect; muduo::CountDownLatch _connect_latch; // 必须先构造 EventLoopThread // 后面的 _baseloop 才能通过 startLoop() 得到有效指针 muduo::net::EventLoopThread _loopthread; muduo::net::EventLoop *_baseloop; muduo::net::TcpClient _client; muduo::net::TcpConnectionPtr _conn; };4.4DictClient.cpp#include DictClient.h #include muduo/net/InetAddress.h #include functional #include iostream DictClient::DictClient( const std::string ip, int port, bool wait_connect) : _wait_connect(wait_connect), _connect_latch(1), _baseloop(_loopthread.startLoop()), _client( _baseloop, muduo::net::InetAddress(ip, port), DictClient) { _client.setConnectionCallback( std::bind( DictClient::onConnection, this, std::placeholders::_1)); _client.setMessageCallback( std::bind( DictClient::onMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)); } void DictClient::connect() { _client.connect(); if (_wait_connect) { // 等待 onConnection() 保存好 _conn _connect_latch.wait(); } } void DictClient::shutdown() { _client.disconnect(); } void DictClient::translate( const std::string word) { if (!_conn || !_conn-connected()) { std::cerr 连接尚未建立 std::endl; return; } std::string request word; request.push_back(\n); _conn-send(request); } void DictClient::onConnection( const muduo::net::TcpConnectionPtr conn) { if (conn-connected()) { _conn conn; if (_wait_connect) { // 连接真正建立以后再唤醒 connect() 中等待的主线程 _connect_latch.countDown(); } std::cout CONNECT SERVER SUCCESS std::endl; return; } _conn.reset(); std::cout CONNECT SHUTDOWN std::endl; } void DictClient::onMessage( const muduo::net::TcpConnectionPtr conn, muduo::net::Buffer *buf, muduo::Timestamp timestamp) { (void)conn; (void)timestamp; while (true) { const char *eol buf-findEOL(); if (eol nullptr) { break; } std::string response( buf-peek(), eol); buf-retrieveUntil(eol 1); if (!response.empty() response.back() \r) { response.pop_back(); } std::cout 翻译结果 response std::endl; } } int main() { DictClient client( 127.0.0.1, 8080); client.connect(); std::string word; while (true) { std::cout 请输入 std::flush; if (!std::getline( std::cin, word)) { break; } if (word quit) { break; } if (word.empty()) { continue; } client.translate(word); } client.shutdown(); return 0; }这里主线程负责读取输入后台 EventLoop 线程负责网络事件。所以服务端响应回来以后真正打印结果的是DictClient::onMessage()而不是发送请求的translate()。这也是后面理解 RPC 异步请求非常重要的一步请求发出去和响应回来本来就是两个不同时间点发生的事件。五、编译和运行下面给 Demo 写一个简单 Makefile。这里的 Muduo 路径按照我当前项目的目录写大家要根据自己的实际安装位置调整。all:DictClient DictServer DictClient:DictClient.cpp g -stdc11 $^ -o $ \ -I ../../muduo \ -L ../../build/release-cpp11/lib -lmuduo_net -lmuduo_base -lpthread DictServer:DictServer.cpp g -stdc11 $^ -o $ \ -I ../../muduo \ -L ../../build/release-cpp11/lib -lmuduo_net -lmuduo_base -lpthread .PHONY:clean clean: rm -rf DictClient DictServer编译make先启动服务端./DictServer再打开另一个终端启动客户端./DictClient可以输入Hello World 你好 abc对应会陆续收到你好 世界 Hello Unknown因为主线程一直在读取输入而响应通过后台 EventLoop 中的onMessage()异步到达所以终端里有时会出现请输入Hello 请输入翻译结果你好或者提示符和响应交错。这不是 RPC 协议出错而是两个线程同时向终端输出造成的显示现象。写在最后到这里我们已经用 Muduo 跑通了一次完整的客户端输入;TCP 发送;服务端 Buffer;消息回调;业务处理;TcpConnection 返回;客户端消息回调.这个词典 Demo 本身不会进入最终 Json-Rpc 框架它的作用是让我们先熟悉 Muduo 的核心对象和回调模型。在当前项目的正式代码里这些能力会继续被封装成MuduoBuffer、MuduoConnection、MuduoServer和MuduoClient让上层协议和 RPC 逻辑不必直接依赖 Muduo 的具体 API。下一步我们就可以开始把“直接操作 Muduo”继续抽象一层为后面的消息协议和 RPC 调用做准备。