muduo源码分析之实现TCP网络库（连接的接收和关闭）

短命女 2022-07-14 05:29 128阅读 0赞

在`EventLoop`、`Channel`、`Poller`三个类中完成对一般描述符、事件循环（poll）的封装。实现了Reactor的基本功能，接下来则需要将网络套接字描述符、I/O函数、等进行封装。

## <table>   <tbody>    <tr>     <td><font><strong>1.传统的TcpServer</strong></font></td>    </tr>   </tbody>  </table> ##

在进行封装之前需要明确我们需要封装的内容有哪些？复习最简单的TcpServer大概需要经历如下步骤：

listenfd= socket(AF_INET,SOCK_STREAM,0);
    strcut sockaddr_in servaddr;
    bind(listenfd,(SA*)&servaddr,sizeof(servaddr));
    listen(listenfd,LISTENQ);
    struct pollfd fds[1024];
    fds[0].fd=listenfd;
    fds[0].events=POLLIN;
    while(1)
    {
        ret=poll(fds,fds.size(),timeout);
        //handle_event
        if(fds[0].revents&POLLIN)
        {
        //new connection
            connectedfd = accept(listenfd,clienAddr,sizeof(clienAddr));
        //add connectedfd into POLL
        }
    }

这是一个比较典型的reactor TcpServer，poll调用的部分已经封装，目前来看至少需要封装的内容有：  
1.网络套接字描述符fd  
2.描述地址结构体`strcut sockaddr_in`  
3.socket相关的系统调用

## <table>   <tbody>    <tr>     <td><font><strong>2.socket操作封装</strong></font></td>    </tr>   </tbody>  </table> ##

### **Endian.h** ###

封装了字节序转换函数（全局函数，位于muduo::net::sockets名称空间中）。

### **SocketsOps.h/ SocketsOps.cc** ###

封装了socket相关系统调用（全局函数，位于muduo::net::sockets名称空间中）。

### **Socket.h/Socket.cc（Socket类）** ###

用RAII方法封装socket file descriptor，包含操作：`listen`、`bind`、`accept`这些操作将调用上述封装的内容。

### **InetAddress.h/InetAddress.cc（InetAddress类）** ###

网际地址sockaddr\_in封装

## <table>   <tbody>    <tr>     <td><font><strong>3.使用Acceptor类封装监听描述符</strong></font></td>    </tr>   </tbody>  </table> ##

这个类用于处理`accept`调用，事实上是对监听套接字`Listenfd`的封装。

Acceptor::Acceptor(EventLoop* loop, const InetAddress& listenAddr)
      : loop_(loop),
        acceptSocket_(sockets::createNonblockingOrDie()),//create socket
        acceptChannel_(loop, acceptSocket_.fd()),//Channel construct
        listenning_(false)
    {
      acceptSocket_.setReuseAddr(true);//不用等待Time_Wait状态结束
      acceptSocket_.bindAddress(listenAddr);//bind
      acceptChannel_.setReadCallback(
          boost::bind(&Acceptor::handleRead, this));
    }

从Acceptor的构造函数可以看出它包含了Socket对象，也就是说它本质上是一个描述符。  
通过Channel类型的对象，设置了当Acceptor所管理的描述符可读时，进行所执行的回调`handleRead`。  
handleRead函数：

void Acceptor::handleRead()
    {
      loop_->assertInLoopThread();
      InetAddress peerAddr(0);
      //FIXME loop until no more
      int connfd = acceptSocket_.accept(&peerAddr);//accpet
      if (connfd >= 0) {
        if (newConnectionCallback_) {
          newConnectionCallback_(connfd, peerAddr);//用户回调
        } else {
          sockets::close(connfd);
        }
      }
    }

当listenfd可读时，说明有新连接，这个时候调用accept创建已连接套接字，并且执行相应的用户回调。  
到目前为止，完成了对监听套接字的简单封装，这个封装是不完全的，因为事实上，当有了TcpServer的概念后，监听套接字对于用户来说应该是不可见的。

不过在对TcpServer进行封装之前，可以测试Acceptor的功能：

void newConnection(int sockfd, const muduo::InetAddress& peerAddr)
    {
      printf("newConnection(): accepted a new connection from %s\n",
             peerAddr.toHostPort().c_str());
      ::write(sockfd, "How are you?\n", 13);
      muduo::sockets::close(sockfd);
    }
    
    int main()
    {
      printf("main(): pid = %d\n", getpid());
    
      muduo::InetAddress listenAddr(9981);
    
      muduo::EventLoop loop;
    
      muduo::Acceptor acceptor(&loop, listenAddr);
    
      acceptor.setNewConnectionCallback(newConnection);
    
      acceptor.listen();
    
      loop.loop();
    }

## <table>   <tbody>    <tr>     <td><font><strong>4.TcpServer接收新的连接</strong></font></td>    </tr>   </tbody>  </table> ##

看了上述最后一个例子我们发现，监听套接字的封装暴露给了用户。所以设计的TcpServer类就需要对Acceptor再封装一层了。

TcpServer::TcpServer(EventLoop* loop, const InetAddress& listenAddr)
      : loop_(CHECK_NOTNULL(loop)),
        name_(listenAddr.toHostPort()),//
        acceptor_(new Acceptor(loop, listenAddr)),//Accept的封装
        started_(false),
        nextConnId_(1)//记录连接数，当有新连接的时候会自增
    {
      acceptor_->setNewConnectionCallback(
          boost::bind(&TcpServer::newConnection, this, _1, _2));
    }

我们看在之前的Acceptor中，只是在main中设置了用户回调，向已连接套接字write了一段文本，便close掉了，对已连接套接字并没有处理。  
现在在Acceptor的回调中设置`TcpServer::newConnection`。

void TcpServer::newConnection(int sockfd, const InetAddress& peerAddr)
    {
        //sockfd 是已连接套接字
      loop_->assertInLoopThread();
      char buf[32];
      snprintf(buf, sizeof buf, "#%d", nextConnId_);
      ++nextConnId_;//连接数+1
      std::string connName = name_ + buf;
    
      LOG_INFO << "TcpServer::newConnection [" << name_
               << "] - new connection [" << connName
               << "] from " << peerAddr.toHostPort();
      //打印特定消息
    
      InetAddress localAddr(sockets::getLocalAddr(sockfd));
    
      //TcpConnection类用来管理已连接套接字
      TcpConnectionPtr conn(
          new TcpConnection(loop_, //EventLoop 
          connName,//ConnectionName
          sockfd, //accepted fd
          localAddr,//
          peerAddr));//
    
      connections_[connName] = conn;//map
      conn->setConnectionCallback(connectionCallback_);//设置用户回调
      conn->setMessageCallback(messageCallback_);//设置用户回调
      conn->connectEstablished();//在这个函数中，会执行onConnection的用户回调
    }

在这个回调中，创建了一个关键的对象，`TcpConnection`，并用`shared_ptr`管理，该Class用来管理已连接套接字。

TcpConnection::TcpConnection(EventLoop* loop,
                                 const std::string& nameArg,
                                 int sockfd,
                                 const InetAddress& localAddr,
                                 const InetAddress& peerAddr)
      : loop_(CHECK_NOTNULL(loop)),//EventLoop
        name_(nameArg),//
        state_(kConnecting),
        socket_(new Socket(sockfd)),//已连接套接字
        channel_(new Channel(loop, sockfd)),//Channel
        localAddr_(localAddr),//
        peerAddr_(peerAddr)//
    {
      LOG_DEBUG << "TcpConnection::ctor[" <<  name_ << "] at " << this
                << " fd=" << sockfd;
      channel_->setReadCallback(
          boost::bind(&TcpConnection::handleRead, this));//当已连接套接字可读时
    }

当已连接套接字可读将回调`TcpConnection::handleRead`

void TcpConnection::handleRead()
    {
      char buf[65536];
      ssize_t n = ::read(channel_->fd(), buf, sizeof buf);//read
      messageCallback_(shared_from_this(), buf, n);//onMessage用户回调
      // FIXME: close connection if n == 0
    }

### **busy loop事件** ###

对于运行`s05/test8.cc`它是一个discard服务，当客户端关闭连接，将进入busy loop，原因在与当客户端断开连接时，客户端发送一个FIN段，服务端返回一个ACK，当服务端TCP收到FIN时，read就返回0（表示EOF），而此时这种状态总是readable，将会不停触发poll返回，就出现busy loop事件了。

## <table>   <tbody>    <tr>     <td><font><strong>5.TcpConnection关闭连接</strong></font></td>    </tr>   </tbody>  </table> ##

关闭连接相对于建立连接要麻烦，因为需要考虑`TcpConnection`的生命周期。

监听套接字可读事件是POLLIN; 已连接套接字正常可读是POLLIN; 正常可写是POLLOUT; 对等方close/shutdown关闭连接，已连接套接字可读是POLLIN | POLLHUP;

在TcpConnection 构造函数中再添加：

// 连接关闭，回调TcpConnection::handleClose
    channel_->setCloseCallback(
        boost::bind(&TcpConnection::handleClose, this));
    // 发生错误，回调TcpConnection::handleError
    channel_->setErrorCallback(
        boost::bind(&TcpConnection::handleError, this));

在 TcpServer::newConnection() 中再添加：

void TcpServer::newConnection(int sockfd, const InetAddress &peerAddr)
    {
        .....
        conn->setCloseCallback(
            boost::bind(&TcpServer::removeConnection, this, _1));
    }

在TcpConnection::handleRead() 中再添加：

void TcpConnection::handleRead()
    {
      char buf[65536];
      ssize_t n = ::read(channel_->fd(), buf, sizeof buf);
      if (n > 0) {
        messageCallback_(shared_from_this(), buf, n);
      } else if (n == 0) {
        handleClose();
      } else {
        handleError();
      }
    }

在来看看`TcpConnection::handleClose()`函数

void TcpConnection::handleClose()
    {
      loop_->assertInLoopThread();
      LOG_TRACE << "TcpConnection::handleClose state = " << state_;
      assert(state_ == kConnected);
      // we don't close fd, leave it to dtor, so we can find leaks easily.
      channel_->disableAll();
      // must be the last line
      closeCallback_(shared_from_this());
    }

回调来回调去的有点晕，整理一下整个过程。  
当已连接套接字发生可读事件，poll返回，将调用调用`Channel::handleEvent()`处理活动通道，调用`TcpConnection::handleRead()`，::read() 返回0，进而调用`TcpConnection::handleClose()`

在`handleClose()`函数中，调用`TcpConnection`的closeCallback\_，这个回调函数是在`TcpServer`里面设定：

conn->setCloseCallback(
            boost::bind(&TcpServer::removeConnection, this, _1));

进而调用`TcpServer::removeConnection`

void TcpServer::removeConnection(const TcpConnectionPtr& conn)
    {
      loop_->assertInLoopThread();
      LOG_INFO << "TcpServer::removeConnection [" << name_
               << "] - connection " << conn->name();
      size_t n = connections_.erase(conn->name());
      assert(n == 1); (void)n;
      loop_->queueInLoop(
          boost::bind(&TcpConnection::connectDestroyed, conn));
    }

最后将调用`TcpConnection::connectDestroy`

void TcpConnection::connectDestroyed()
    {
      loop_->assertInLoopThread();
      assert(state_ == kConnected);
      setState(kDisconnected);
      channel_->disableAll();
      connectionCallback_(shared_from_this());//回调用户onConnection Callback
    
      loop_->removeChannel(get_pointer(channel_));//Poll不再关注此通道
    }

## <table>   <tbody>    <tr>     <td><font><strong>6.参考</strong></font></td>    </tr>   </tbody>  </table> ##

1.Linux多线程服务端编程 使用muduo C++网络库  
2.[http://blog.csdn.net/jnu\_simba/article/details/14556387][http_blog.csdn.net_jnu_simba_article_details_14556387]  
3.unp1

[http_blog.csdn.net_jnu_simba_article_details_14556387]: http://blog.csdn.net/jnu_simba/article/details/14556387