484-手写C++muduo库(TcpServer)

「爱情、让人受尽委屈。」 2022-09-08 14:56 195阅读 0赞

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBALeael-azveWuhw_size_20_color_FFFFFF_t_70_g_se_x_16]

## Callbacks.h ##

#pragma once
    
    #include <memory>
    #include <functional>
    
    class Buffer;
    class TcpConnection;
    class Timestamp;
    
    using TcpConnectionPtr = std::shared_ptr<TcpConnection>;
    using ConnectionCallback = std::function<void (const TcpConnectionPtr&)>;
    using CloseCallback = std::function<void (const TcpConnectionPtr&)>;
    using WriteCompleteCallback = std::function<void (const TcpConnectionPtr&)>;
    using MessageCallback = std::function<void (const TcpConnectionPtr&,
                                            Buffer*,
                                            Timestamp)>;
    using HighWaterMarkCallback = std::function<void (const TcpConnectionPtr&, size_t)>;

## TcpServer.h ##

#pragma once
    
    /**
     * 用户使用muduo编写服务器程序
     */ 
    #include "EventLoop.h"
    #include "Acceptor.h"
    #include "InetAddress.h"
    #include "noncopyable.h"
    #include "EventLoopThreadPool.h"
    #include "Callbacks.h"
    #include "TcpConnection.h"
    #include "Buffer.h"
    
    #include <functional>
    #include <string>
    #include <memory>
    #include <atomic>
    #include <unordered_map>//哈希表 
    
    //我们把需要用到的头文件都包含在这里，方便用户使用 
    //对外的服务器编程使用的类
    class TcpServer : noncopyable
    {
        
    public:
        using ThreadInitCallback = std::function<void(EventLoop*)>;
    
        enum Option//选项，是否对端口可重用 
        {
        
            kNoReusePort,
            kReusePort,
        };
    
        TcpServer(EventLoop *loop,
                    const InetAddress &listenAddr,
                    const std::string &nameArg,
                    Option option = kNoReusePort);
        ~TcpServer();
        
        //用户在使用muduo库必须设置的 
        void setThreadInitcallback(const ThreadInitCallback &cb) {
         threadInitCallback_ = cb; }
        void setConnectionCallback(const ConnectionCallback &cb) {
         connectionCallback_ = cb; }
        void setMessageCallback(const MessageCallback &cb) {
         messageCallback_ = cb; }
        void setWriteCompleteCallback(const WriteCompleteCallback &cb) {
         writeCompleteCallback_ = cb; }
    
        //设置底层subloop的个数
        void setThreadNum(int numThreads);
    
        //开启服务器监听 实际上就是开启mainloop的accptor的listen 
        void start();
    private:
    	//私有的内部使用的接口 
        void newConnection(int sockfd, const InetAddress &peerAddr);//有新连接来了 
        void removeConnection(const TcpConnectionPtr &conn);//有连接断开了，不要这条连接了 
        void removeConnectionInLoop(const TcpConnectionPtr &conn);
    
        using ConnectionMap = std::unordered_map<std::string, TcpConnectionPtr>;
    
        EventLoop *loop_;//baseLoop 用户定义的loop 一个线程一个loop循环 
    
        const std::string ipPort_;//服务器的IP地址端口号 
        const std::string name_;//服务器的名称 
    
        std::unique_ptr<Acceptor> acceptor_;//运行在mainLoop，任务就是监听新连接事件
    
        std::shared_ptr<EventLoopThreadPool> threadPool_;//线程池 one loop per thread
    
        ConnectionCallback connectionCallback_;//有新连接时的回调
        MessageCallback messageCallback_;//已连接用户有读写消息时的回调 reactor调用 
        WriteCompleteCallback writeCompleteCallback_;//消息发送完成以后的回调
    
        ThreadInitCallback threadInitCallback_;//loop线程初始化的回调
    
        std::atomic_int started_;//标志 
    
        int nextConnId_;
        ConnectionMap connections_;//保存所有的连接
    };

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBALeael-azveWuhw_size_20_color_FFFFFF_t_70_g_se_x_16 1]  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBALeael-azveWuhw_size_20_color_FFFFFF_t_70_g_se_x_16 2]  
![在这里插入图片描述][c057df7ff7754358aa33ccd7f0ecd663.png]  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBALeael-azveWuhw_size_20_color_FFFFFF_t_70_g_se_x_16 3]  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBALeael-azveWuhw_size_20_color_FFFFFF_t_70_g_se_x_16 4]

## TcpServer.cc ##

#include "TcpServer.h"
    #include "Logger.h"
    #include "TcpConnection.h"
    
    #include <strings.h>
    #include <functional>
    
    static EventLoop* CheckLoopNotNull(EventLoop *loop)
    {
        
        if (loop == nullptr)
        {
        
            LOG_FATAL("%s:%s:%d mainLoop is null! \n", __FILE__, __FUNCTION__, __LINE__);
        }
        return loop;
    }
    
    TcpServer::TcpServer(EventLoop *loop,
                    const InetAddress &listenAddr,
                    const std::string &nameArg,
                    Option option)
                    : loop_(CheckLoopNotNull(loop))//不能为空 
                    , ipPort_(listenAddr.toIpPort())
                    , name_(nameArg)
                    , acceptor_(new Acceptor(loop, listenAddr, option == kReusePort))
                    , threadPool_(new EventLoopThreadPool(loop, name_))
                    , connectionCallback_()
                    , messageCallback_()
                    , nextConnId_(1)
                    , started_(0)
    {
        
        //当有新用户连接时，会执行TcpServer::newConnection回调
        acceptor_->setNewConnectionCallback(std::bind(&TcpServer::newConnection, this, 
            std::placeholders::_1, std::placeholders::_2));
    }
    
    TcpServer::~TcpServer()
    {
        
        for (auto &item : connections_)
        {
        
            //这个局部的shared_ptr智能指针对象，出右括号，可以自动释放new出来的TcpConnection对象资源了
            TcpConnectionPtr conn(item.second); 
            item.second.reset();
    
            //销毁连接
            conn->getLoop()->runInLoop(
                std::bind(&TcpConnection::connectDestroyed, conn)
            );
        }
    }
    
    //设置底层subloop的个数
    void TcpServer::setThreadNum(int numThreads)
    {
        
        threadPool_->setThreadNum(numThreads);
    }
    
    //开启服务器监听   loop.loop()
    void TcpServer::start()
    {
        
        if (started_++ == 0)//防止一个TcpServer对象被start多次
        {
        
            threadPool_->start(threadInitCallback_);//启动底层的loop线程池
            loop_->runInLoop(std::bind(&Acceptor::listen, acceptor_.get()));
        }
    }
    
    //有一个新的客户端的连接，acceptor会执行这个回调操作
    void TcpServer::newConnection(int sockfd, const InetAddress &peerAddr)
    {
        
        //轮询算法，选择一个subLoop，来管理channel
        EventLoop *ioLoop = threadPool_->getNextLoop(); 
        char buf[64] = {
        0};
        snprintf(buf, sizeof buf, "-%s#%d", ipPort_.c_str(), nextConnId_);
        ++nextConnId_;
        std::string connName = name_ + buf;
    
        LOG_INFO("TcpServer::newConnection [%s] - new connection [%s] from %s \n",
            name_.c_str(), connName.c_str(), peerAddr.toIpPort().c_str());
    
        //通过sockfd获取其绑定的本机的ip地址和端口信息
        sockaddr_in local;
        ::bzero(&local, sizeof local);
        socklen_t addrlen = sizeof local;
        if (::getsockname(sockfd, (sockaddr*)&local, &addrlen) < 0)
        {
        
            LOG_ERROR("sockets::getLocalAddr");
        }
        InetAddress localAddr(local);
    
        //根据连接成功的sockfd，创建TcpConnection连接对象
        TcpConnectionPtr conn(new TcpConnection(
                                ioLoop,
                                connName,
                                sockfd,   // Socket Channel
                                localAddr,
                                peerAddr));
        connections_[connName] = conn;
        //下面的回调都是用户设置给TcpServer=>TcpConnection=>Channel=>Poller=>notify channel调用回调
        conn->setConnectionCallback(connectionCallback_);
        conn->setMessageCallback(messageCallback_);
        conn->setWriteCompleteCallback(writeCompleteCallback_);
    
        //设置了如何关闭连接的回调   conn->shutDown()
        conn->setCloseCallback(
            std::bind(&TcpServer::removeConnection, this, std::placeholders::_1)
        );
    
        //直接调用TcpConnection::connectEstablished
        ioLoop->runInLoop(std::bind(&TcpConnection::connectEstablished, conn));
    }
    
    void TcpServer::removeConnection(const TcpConnectionPtr &conn)
    {
        
        loop_->runInLoop(
            std::bind(&TcpServer::removeConnectionInLoop, this, conn)
        );
    }
    
    void TcpServer::removeConnectionInLoop(const TcpConnectionPtr &conn)
    {
        
        LOG_INFO("TcpServer::removeConnectionInLoop [%s] - connection %s\n", 
            name_.c_str(), conn->name().c_str());
    
        connections_.erase(conn->name());
        EventLoop *ioLoop = conn->getLoop(); 
        ioLoop->queueInLoop(
            std::bind(&TcpConnection::connectDestroyed, conn)
        );
    }

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBALeael-azveWuhw_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/7dfbd1753b774543b062f158a2d03eb7.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBALeael-azveWuhw_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220829/f5968c24ac6a4ec8a085ccba5bc10ccb.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBALeael-azveWuhw_size_20_color_FFFFFF_t_70_g_se_x_16 2]: /images/20220829/67868911098346dd98b94b027b166832.png
[c057df7ff7754358aa33ccd7f0ecd663.png]: /images/20220829/045b7bca9d134202889f1c94efaed585.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBALeael-azveWuhw_size_20_color_FFFFFF_t_70_g_se_x_16 3]: /images/20220829/767d4a25bffd48c1be0c2abec34786f8.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBALeael-azveWuhw_size_20_color_FFFFFF_t_70_g_se_x_16 4]: /images/20220829/49bf792c512442fbb276c966ca5d5439.png