[集群聊天项目] muduo网络库

news2025/1/11 20:00:00

网络服务器编程常用模型

【方案1】： accept + read/write
不是并发服务器

【方案2】： accept + fork - process-pre-connection
适合并发连接数不大，计算任务工作量大于fork的开销

【方案3】：accept + thread thread-pre-connection
比方案2的开销小了一点，但是并发造成线程堆积过多

【方案4】： muduo的网络设计：reactors in threads - one loop per thread

【方案5】： reactors in process - one loop pre process
nginx服务器的网络模块设计，基于进程设计，采用多个Reactors充当I/O进程和工作进程，通过一把accept锁，完美解决多个Reactors的“惊群现象”

什么是muduo网络库

高并发，因此服务器要用muduo编程
在这里插入图片描述

方案的特点是one loop per thread（一个线程一个时间循环），有一个main reactor负载accept连接（I/O线程），然后把连接分发到某个sub reactor（采用round-robin[轮询]的方式来选择sub reactor），该连接的所用操作都在那个sub reactor所处的线程中完成，注册到某个线程的epoll事件上，那么所做的所有读写操作都在这个epoll上完成。多个连接可能被分派到多个线程中，以充分利用CPU。

Reactor poll的大小是固定的，根据CPU的数目确定。

如果有过多的耗费CPU I/O的计算任务，可以提交到创建的ThreadPool线程池中专门处理耗时的计算任
务。

// 设置EventLoop的线程个数，底层通过EventLoopThreadPool线程池管理线程类EventLoopThread
_server.setThreadNum(10);

一个Base IO thread负责accept新的连接，接收到新的连接以后，使用轮询的方式在reactor pool中找
到合适的sub reactor将这个连接挂载上去，这个连接上的所有任务都在这个sub reactor上完成。

什么是epoll

学习资料
epoll 是 Linux 内核的可扩展 I/O 事件通知机制，其最大的特点就是性能优异。
在这里插入图片描述

muduo网络库服务器编程

moduo库的使用需要链接libmuduo_base.so、libmuduo_net.so、libpthread.so库，我们执行的时候需要-lmuduo_net -lmuduo_base -lpthread，需要配置这些。

muduo网络库给用户提供了两个主要的类

TcpSever : 用于编写服务器程序的
TcpClient : 用于编写客户端程序的

eopll + 线程池
好处：能把网络I/O的代码和业务代码区分开

业务代码的暴露主要有两个：用户的连接和断开，用户的可读写事件。我们只需要关注这两个事件怎么做，至于什么时候发生这些事情（网络库上报）以及怎么监听这些事件的发生，都是由网络库实现好的。

基于muduo网络库开发服务器程序

1. 组合TcpSevrer对象：不指定构造TcpServer _server;就要用默认构造，但是进入到TcpServer.h中发现TcpServer没有默认构造，因此要指定_server相应的构造，在pubilc中。不指定，ChatServer类无法创建对象。

2. 创建EventLoop事件循环对象的指针

TcpServer _server; // #1
EventLoop *_loop;  // #2 epoll

3. 明确TcpServer构造函数需要什么函数，输出ChatServer的构造函数

ChatServer(EventLoop *loop,               // 事件循环
       const InetAddress &listenAddr, // IP地址+端口
           const string &nameArg)         // 服务器的名字
    : _server(loop, listenAddr, nameArg), _loop(loop)
{}

4.在当前服务器类的构造函数中，注册处理连接的回调函数和处理读写事件的回调函数。将处理业务的代码onConnection和onMessage分离出来，和对应的Callback代码绑定到一起，其中onConnection和onMessage两个函数的参数类型和数量是根据Callback函数设置的。
在这里插入图片描述

// 给服务器注册用户连接的创建和断开回调，一旦监听到此活动，就会调用onConnection函数
// 人家的Callback函数只有一个参数，但是自己写的onConnection有两个参数，隐含一个this
// _1是参数占位符（需要using namespace placeholders;），是onConnection括号中的参数
_server.setConnectionCallback(std::bind(&ChatServer::onConnection, this, _1));
// 给服务器注册用户读写事件回调
_server.setMessageCallback(std::bind(&ChatServer::onMessage, this, _1, _2, _3));
// **************************************************************
// 专门处理用户的连接创建和断开 epoll  listenfd accept
void onConnection(const TcpConnectionPtr &conn)
{// 放业务代码}

// 专门处理用户的读写事件
void onMessage(const TcpConnectionPtr &conn, // 连接
               Buffer *buffer,               // 缓冲区
               Timestamp time)               // 接收到数据的时间信息
{// 放业务代码}

5. 设置合适的服务端线程数，muduo库会自己分配I/O线程和worker线程

// 设置服务器端的线程数量 1个I/O线程 3个worker线程
_server.setThreadNum(4);

全部代码：

#include <muduo/net/TcpServer.h>
#include <muduo/net/TcpClient.h>
#include <iostream>
#include <functional> //绑定器在此文件中
#include <string>
using namespace std;
using namespace muduo;
using namespace muduo::net;
using namespace placeholders;

class ChatServer
{
public:
    // #3
    ChatServer(EventLoop *loop,
               const InetAddress &listenAddr,
               const string &nameArg)
        : _server(loop, listenAddr, nameArg), _loop(loop)
    {
        // #4
        _server.setConnectionCallback(std::bind(&ChatServer::onConnection, this, _1));
        _server.setMessageCallback(std::bind(&ChatServer::onMessage, this, _1, _2, _3));

        // #5
        _server.setThreadNum(4);
    }

    // 开启事件循环
    void start()
    {
        _server.start();
    }

private:
    // 专门处理用户的连接创建和断开 epoll  listenfd accept
    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();
        }
    }

    // 专门处理用户的读写事件
    void onMessage(const TcpConnectionPtr &conn,
                   Buffer *buffer,
                   Timestamp time)
    {
        // 放业务代码
        string buff = buffer->retrieveAllAsString();
        cout << "recv data:" << buff << "time: " << time.toString() << endl;
        conn->send(buff);
    }

    TcpServer _server;
    EventLoop *_loop;
};

int main()
{
    EventLoop loop; // epoll
    InetAddress addr("127.0.0.1", 6000);
    ChatServer server(&loop, addr, "ChatServer");

    server.start(); // 启动服务， listenfd epoll_atl=>epoll
    loop.loop();    // epoll_wait以阻塞方式等待新用户链接，已连接用户的读写事件等操作

    return 0;
}