muduo库分析——net篇（5）Tcp

程序员文章站 2022-06-14 10:22:02

...

已经总结了，TcpServer是一个控制器，维护了一个TcpConnection列表和一个Acceptor

Acceptor作用是建立一个监听套接字，用来接受新的连接，并调用相关回调函数

TcpConnection是拥有Socket和Channel两个成员，Socket用于监听相关连接数据，Channel用于事件管理

TcpServer还有一个EventLoopThreadPool成员，这是事件分发到其他线程，在这里就是讲新的连接给其他线程进行管理和监听数据

TcpServer的EventLoop被Acceptor和EventLoopThreadPool共有，这是因为TcpServer是一个总的控制，它的作用是接受新的连接，分发连接到其他线程这两个功能，因此这个IO线程的作用就是管理Acceptor的新连接事件，开启多个线程，以轮调的方式把每个连接分发给新的线程，这些事件都统一交给EventLoop处理

muduo库分析——net篇（5）Tcp

Acceptor一些注释：

  acceptSocket_.setReuseAddr(true); //设置socket，主要在time_wait情况下，该地址：端口可以立即重用
  acceptSocket_.setReusePort(reuseport);//设置端口可重用，多个线程或进程可以监听此端口
  acceptSocket_.bindAddress(listenAddr);//设置地址簇
  acceptChannel_.setReadCallback(      //设置Channel回调函数
      boost::bind(&Acceptor::handleRead, this));

void Acceptor::handleRead()     //监听描述符一旦有可读事件说明有新的链接到来，设置相应回调函数
{
  loop_->assertInLoopThread();  //判断是否在accpetor线程中
  InetAddress peerAddr;         //链接到来的IP地址器
  //FIXME loop until no more
  int connfd = acceptSocket_.accept(&peerAddr);
  if (connfd >= 0)
  {
    // string hostport = peerAddr.toIpPort();
    // LOG_TRACE << "Accepts of " << hostport;
    if (newConnectionCallback_) //新的链接有相应回调函数
    {
      newConnectionCallback_(connfd, peerAddr);
    }
    else
    {
      sockets::close(connfd);
    }
  }
  else
  {
    LOG_SYSERR << "in Acceptor::handleRead";
    // Read the section named "The special problem of
    // accept()ing when you can't" in libev's doc.
    // By Marc Lehmann, author of libev.
    if (errno == EMFILE)
    {
      ::close(idleFd_);     //关闭空文件描述符，将错误文件描述符转移
      idleFd_ = ::accept(acceptSocket_.fd(), NULL, NULL);
      ::close(idleFd_);     //将错误文件描述符关闭，防止占用文件描述符资源
                            //重新获取新的文件描述符以备不时之需
      idleFd_ = ::open("/dev/null", O_RDONLY | O_CLOEXEC);
    }
  }

这是一个新的连接到来的过程

 ///Acceptor的handleread在accept有新的可读事件时设置相关回调函数
      acceptor_->setNewConnectionCallback(
      boost::bind(&TcpServer::newConnection, this, _1, _2));

///新的连接到来时回调函数
void TcpServer::newConnection(int sockfd, const InetAddress& peerAddr)
{
  loop_->assertInLoopThread();
  EventLoop* ioLoop = threadPool_->getNextLoop();//获得当前空闲的loop
  char buf[64];
  snprintf(buf, sizeof buf, "-%s#%d", ipPort_.c_str(), nextConnId_);
  ++nextConnId_;
  string connName = name_ + buf;

  LOG_INFO << "TcpServer::newConnection [" << name_
           << "] - new connection [" << connName
           << "] from " << peerAddr.toIpPort();
  InetAddress localAddr(sockets::getLocalAddr(sockfd)); //获得本地地址器用于新建TcpConnection
  // FIXME poll with zero timeout to double confirm the new connection
  // FIXME use make_shared if necessary
  TcpConnectionPtr conn(new TcpConnection(ioLoop,
                                          connName,
                                          sockfd,
                                          localAddr,
                                          peerAddr));
  connections_[connName] = conn;
  conn->setConnectionCallback(connectionCallback_); //相关设置
  conn->setMessageCallback(messageCallback_);
  conn->setWriteCompleteCallback(writeCompleteCallback_);
  conn->setCloseCallback(
      boost::bind(&TcpServer::removeConnection, this, _1)); // FIXME: unsafe 应该是因为传递this不安全，而应该传递shared_from_this
  //acceptor接受新的连接后，将其分配给所属io线程进行处理
  ioLoop->runInLoop(boost::bind(&TcpConnection::connectEstablished, conn));
}

这里要注意，分发给新的IO线程后，相关的注册都要交由这个IO线程调用

muduo库分析——net篇（5）Tcp