? ? ? ? ? 今天來學(xué)習(xí)一下muduo源碼中client和server間的大致通信流程，以echo服務(wù)為例，先看一下echo對面的main函數(shù)代碼。

#include "examples/simple/echo/echo.h"#include "muduo/base/Logging.h" #include "muduo/net/EventLoop.h"#include <unistd.h>// using namespace muduo; // using namespace muduo::net;int main() {LOG_INFO << "pid = " << getpid(); //getpid()獲取進程號muduo::net::EventLoop loop;muduo::net::InetAddress listenAddr(2007); //綁定IP地址和端口EchoServer server(&loop, listenAddr);server.start();loop.loop(); }

? ? ? ?先實例化一個loop，這會調(diào)用loop的構(gòu)造函數(shù)，看一下loop的構(gòu)造函數(shù)。

EventLoop::EventLoop(): looping_(false), //判斷是否在loopquit_(false), //判斷是否退出的標(biāo)志eventHandling_(false), //處理handevent的標(biāo)志callingPendingFunctors_(false), //判斷當(dāng)前是不是在執(zhí)行方法隊列iteration_(0),threadId_(CurrentThread::tid()), //當(dāng)前線程IDpoller_(Poller::newDefaultPoller(this)), //創(chuàng)建一個 poll 或 epoll 對象timerQueue_(new TimerQueue(this)), //創(chuàng)建一個計時器wakeupFd_(createEventfd()), //發(fā)送喚醒loop消息的描述符,隨便寫點消息即可喚醒wakeupChannel_(new Channel(this, wakeupFd_)), //wakeupChannel_用來自己給自己通知的一個通道，該通道會納入到poller來管理currentActiveChannel_(NULL) //當(dāng)前活躍的channel鏈表指針 {LOG_DEBUG << "EventLoop created " << this << " in thread " << threadId_;if (t_loopInThisThread) //判斷是否是本線程的loop,是一個loop類型的指針{LOG_FATAL << "Another EventLoop " << t_loopInThisThread<< " exists in this thread " << threadId_; //用LOG_FATAL終止abort它}else{t_loopInThisThread = this; //this賦給線程局部數(shù)據(jù)指針}//設(shè)定wakeupChannel的回調(diào)函數(shù)，即EventLoop自己的的handleRead函數(shù)wakeupChannel_->setReadCallback(std::bind(&EventLoop::handleRead, this)); //channel->handleEventWithGuard會調(diào)用到handleRead// we are always reading the wakeupfdwakeupChannel_->enableReading(); //注冊wakeupFd_到poller }

? ? ?loop就相當(dāng)于一個reactor，loop構(gòu)造函數(shù)new了epoll，創(chuàng)建了喚醒描述符，并注冊到epoll。接著對EchoServer進行了構(gòu)造，調(diào)用了start函數(shù)。

EchoServer::EchoServer(muduo::net::EventLoop* loop,const muduo::net::InetAddress& listenAddr): server_(loop, listenAddr, "EchoServer") {server_.setConnectionCallback(std::bind(&EchoServer::onConnection, this, _1)); //連接到達server_.setMessageCallback(std::bind(&EchoServer::onMessage, this, _1, _2, _3)); //消息到達 }void EchoServer::start() {server_.start(); }

? ? ??EchoServer構(gòu)造函數(shù)中設(shè)置了連接達到和消息達到的回調(diào)函數(shù)，并構(gòu)建了TcpServer。再看TcpServer的構(gòu)造函數(shù)。

TcpServer::TcpServer(EventLoop* loop,const InetAddress& listenAddr,const string& nameArg,Option option): loop_(CHECK_NOTNULL(loop)),ipPort_(listenAddr.toIpPort()),name_(nameArg),acceptor_(new Acceptor(loop, listenAddr, option == kReusePort)),threadPool_(new EventLoopThreadPool(loop, name_)),connectionCallback_(defaultConnectionCallback),messageCallback_(defaultMessageCallback),nextConnId_(1) { /* * 設(shè)置回調(diào)函數(shù)，當(dāng)有客戶端請求時，Acceptor接收客戶端請求，然后調(diào)用這里設(shè)置的回調(diào)函數(shù)* 回調(diào)函數(shù)用于創(chuàng)建TcpConnection連接*/acceptor_->setNewConnectionCallback(std::bind(&TcpServer::newConnection, this, _1, _2)); }

? ? ??TcpServer構(gòu)造函數(shù)new了一個acceptor_，設(shè)置了當(dāng)有新連接達到時該調(diào)用的回調(diào)函數(shù)。看一下acceptor的構(gòu)造函數(shù)。

Acceptor::Acceptor(EventLoop* loop, const InetAddress& listenAddr, bool reuseport): loop_(loop),acceptSocket_(sockets::createNonblockingOrDie(listenAddr.family())),//初始化創(chuàng)建sockt fdacceptChannel_(loop, acceptSocket_.fd()),//初始化channellistenning_(false),idleFd_(::open("/dev/null", O_RDONLY | O_CLOEXEC)) {assert(idleFd_ >= 0);acceptSocket_.setReuseAddr(true);acceptSocket_.setReusePort(reuseport);acceptSocket_.bindAddress(listenAddr);acceptChannel_.setReadCallback(std::bind(&Acceptor::handleRead, this));// //有可讀事件,當(dāng)fd可讀時調(diào)用回調(diào)函數(shù)hanleRead }

? ? ?Acceptor構(gòu)造函數(shù)中設(shè)置了當(dāng)有可讀事件是該調(diào)用的構(gòu)造函數(shù)，然后start()函數(shù)和loop()函數(shù)開始調(diào)用。

void TcpServer::start() {if (started_.getAndSet(1) == 0){threadPool_->start(threadInitCallback_);//啟動線程池,threadInitCallback_創(chuàng)建好所有線程后調(diào)用的回調(diào)函數(shù)assert(!acceptor_->listenning());loop_->runInLoop( //直接調(diào)用linsten函數(shù)std::bind(&Acceptor::listen, get_pointer(acceptor_)));} } void EventLoop::loop() {assert(!looping_);assertInLoopThread(); //事件循環(huán)必須在IO線程中，即創(chuàng)建該evenloop的線程looping_ = true;quit_ = false; // FIXME: what if someone calls quit() before loop() ?LOG_TRACE << "EventLoop " << this << " start looping";while (!quit_){activeChannels_.clear(); //activeChannels_是一個vector等待io復(fù)用函數(shù)返回pollReturnTime_ = poller_->poll(kPollTimeMs, &activeChannels_); //調(diào)用poll返回活動的事件，有可能是喚醒返回的++iteration_;if (Logger::logLevel() <= Logger::TRACE){printActiveChannels();}// TODO sort channel by priority 按優(yōu)先級排序//處理IO事件eventHandling_ = true;for (Channel* channel : activeChannels_) //遍歷通道來進行處理{currentActiveChannel_ = channel;currentActiveChannel_->handleEvent(pollReturnTime_); //pollReturnTime_是poll返回的時刻}currentActiveChannel_ = NULL; //處理完了賦空eventHandling_ = false;//執(zhí)行方法隊列中的方法[方法隊列functors，我們可以跨線程的往里面添加新的方法，這些方法會在處理完io事件后執(zhí)行]doPendingFunctors(); //這個設(shè)計也能夠進行計算任務(wù)} void Acceptor::listen() {loop_->assertInLoopThread(); //保證是在IO線程listenning_ = true;acceptSocket_.listen();acceptChannel_.enableReading(); 注冊可讀事件 }

? ? ? ?start函數(shù)中在loop里面調(diào)用了Acceptor的listen函數(shù)注冊可讀事件(將可讀事件加入epoll)，接著loop函數(shù)在while循環(huán)里面調(diào)用poller_->poll()。接著看一下epoll的poll函數(shù)。

Timestamp EPollPoller::poll(int timeoutMs, ChannelList* activeChannels) {LOG_TRACE << "fd total count " << channels_.size();int numEvents = ::epoll_wait(epollfd_,&*events_.begin(), //events_已初始化static_cast<int>(events_.size()), //監(jiān)控套接字的數(shù)目timeoutMs);int savedErrno = errno;Timestamp now(Timestamp::now());if (numEvents > 0){LOG_TRACE << numEvents << " events happened";fillActiveChannels(numEvents, activeChannels);if (implicit_cast<size_t>(numEvents) == events_.size()) //如果返回的事件數(shù)目等于當(dāng)前事件數(shù)組大小，就分配2倍空間{events_.resize(events_.size()*2);}}else if (numEvents == 0){LOG_TRACE << "nothing happened";}else{// error happens, log uncommon onesif (savedErrno != EINTR){errno = savedErrno;LOG_SYSERR << "EPollPoller::poll()";}}return now; }

? ? ? ?代碼中有調(diào)用epoll_wait函數(shù)。

? ? ? ?當(dāng)有新連接到達時，epoll返回，currentActiveChannel_->handleEvent(上述loop.loop函數(shù)中)被調(diào)用，看一下對應(yīng)代碼。

//處理所有發(fā)生的事件，如果活著，底層調(diào)用handleEventWithGuard void Channel::handleEvent(Timestamp receiveTime) 事件到來調(diào)用handleEvent處理 {std::shared_ptr<void> guard; //守護if (tied_){guard = tie_.lock();if (guard){handleEventWithGuard(receiveTime);}}else{handleEventWithGuard(receiveTime);} }//處理所有發(fā)生的事件 //EPOLLIN ：表示對應(yīng)的文件描述符可以讀； //EPOLLOUT：表示對應(yīng)的文件描述符可以寫； //EPOLLPRI：表示對應(yīng)的文件描述符有緊急的數(shù)據(jù)可讀 //EPOLLERR：表示對應(yīng)的文件描述符發(fā)生錯誤； //EPOLLHUP：表示對應(yīng)的文件描述符被掛斷； //EPOLLET：表示對應(yīng)的文件描述符有事件發(fā)生； void Channel::handleEventWithGuard(Timestamp receiveTime) { eventHandling_ = true; LOG_TRACE << reventsToString();if ((revents_ & POLLHUP) && !(revents_ & POLLIN)) //判斷返回事件類型{if (logHup_){LOG_WARN << "fd = " << fd_ << " Channel::handle_event() POLLHUP";}if (closeCallback_) closeCallback_();}if (revents_ & POLLNVAL) //不合法文件描述符{LOG_WARN << "fd = " << fd_ << " Channel::handle_event() POLLNVAL";}if (revents_ & (POLLERR | POLLNVAL)){if (errorCallback_) errorCallback_();}if (revents_ & (POLLIN | POLLPRI | POLLRDHUP)) //POLLRDHUP是對端關(guān)閉連接事件，如shutdown等{if (readCallback_) readCallback_(receiveTime);}if (revents_ & POLLOUT){if (writeCallback_) writeCallback_();}eventHandling_ = false; }

? ? ? 是可讀事件，readCallback_被調(diào)用，[acceptChannel_.setReadCallback(std::bind(&Acceptor::handleRead, this))]有設(shè)置，則Acceptor::handleRead()函數(shù)被調(diào)用。

void Acceptor::handleRead() //新連接到達由acccptor處理 {loop_->assertInLoopThread();InetAddress peerAddr;//FIXME loop until no moreint connfd = acceptSocket_.accept(&peerAddr);//這里時真正接收連接if (connfd >= 0){// string hostport = peerAddr.toIpPort();// LOG_TRACE << "Accepts of " << hostport;if (newConnectionCallback_){newConnectionCallback_(connfd, peerAddr);//將新連接信息傳送到回調(diào)函數(shù)中}else//沒有回調(diào)函數(shù)則關(guān)閉client對應(yīng)的fd{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);}} }

? ? ?newConnectionCallback_被調(diào)用，[acceptor_->setNewConnectionCallback(std::bind(&TcpServer::newConnection, this, _1, _2));]，則調(diào)用tcpServer中的newConnection函數(shù)。

void TcpServer::newConnection(int sockfd, const InetAddress& peerAddr) {loop_->assertInLoopThread();EventLoop* ioLoop = threadPool_->getNextLoop();//從事件驅(qū)動線程池中取出一個線程給TcpConnection /* 為TcpConnection生成獨一無二的名字 */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();/* * 根據(jù)sockfd獲取tcp連接在本地的<地址，端口>* getsockname(int fd, struct sockaddr*, int *size);*/InetAddress localAddr(sockets::getLocalAddr(sockfd));// FIXME poll with zero timeout to double confirm the new connection// FIXME use make_shared if necessary/* 創(chuàng)建一個新的TcpConnection代表一個Tcp連接 */TcpConnectionPtr conn(new TcpConnection(ioLoop,connName,sockfd,localAddr,peerAddr));/* 添加到所有tcp 連接的map中，鍵是tcp連接獨特的名字（服務(wù)器名+客戶端<地址，端口>） */connections_[connName] = conn;/* 為tcp連接設(shè)置回調(diào)函數(shù)（由用戶提供） */conn->setConnectionCallback(connectionCallback_);conn->setMessageCallback(messageCallback_);conn->setWriteCompleteCallback(writeCompleteCallback_);/* * 關(guān)閉回調(diào)函數(shù)，由TcpServer設(shè)置，作用是將這個關(guān)閉的TcpConnection從map中刪除* 當(dāng)poll返回后，發(fā)現(xiàn)被激活的原因是EPOLLHUP，此時需要關(guān)閉tcp連接* 調(diào)用Channel的CloseCallback，進而調(diào)用TcpConnection的handleClose，進而調(diào)用removeConnection*/conn->setCloseCallback(std::bind(&TcpServer::removeConnection, this, _1)); // FIXME: unsafe/* * 連接建立后，調(diào)用TcpConnection連接建立成功的函數(shù)* 1.新建的TcpConnection所在事件循環(huán)是在事件循環(huán)線程池中的某個線程* 2.所以TcpConnection也就屬于它所在的事件驅(qū)動循環(huán)所在的那個線程* 3.調(diào)用TcpConnection的函數(shù)時也就應(yīng)該在自己所在線程調(diào)用* 4.所以需要調(diào)用runInLoop在自己的那個事件驅(qū)動循環(huán)所在線程調(diào)用這個函數(shù)* 5.當(dāng)前線程是TcpServer的主線程，不是TcpConnection的線程，如果在這個線程直接調(diào)用會阻塞監(jiān)聽客戶端請求* 6.其實這里不是因為線程不安全，即使在這個線程調(diào)用也不會出現(xiàn)線程不安全，因為TcpConnection本就是由這個線程創(chuàng)建的*/ioLoop->runInLoop(std::bind(&TcpConnection::connectEstablished, conn)); }

connectionCallback_和messageCallback_就是echo.cc中設(shè)置的，connectEstablished會調(diào)用了用戶設(shè)置的connectionCallback.

新連接到達之后就new了一個TcpConnection，看一下TcpConnection的構(gòu)造函數(shù)。

TcpConnection::TcpConnection(EventLoop* loop,const string& nameArg,int sockfd,const InetAddress& localAddr,const InetAddress& peerAddr): loop_(CHECK_NOTNULL(loop)),name_(nameArg),state_(kConnecting), //輸入輸出緩沖區(qū)reading_(true),socket_(new Socket(sockfd)), //RAII管理已連接套接字channel_(new Channel(loop, sockfd)), //使用Channel管理套接字上的讀寫localAddr_(localAddr),peerAddr_(peerAddr),highWaterMark_(64*1024*1024) {//設(shè)置事件分發(fā)器的各事件回調(diào) (將TcpConnection類的四個事件處理函數(shù)設(shè)置為事件分發(fā)器對應(yīng)的回調(diào)函數(shù))channel_->setReadCallback(std::bind(&TcpConnection::handleRead, this, _1));channel_->setWriteCallback(std::bind(&TcpConnection::handleWrite, this));channel_->setCloseCallback(std::bind(&TcpConnection::handleClose, this));channel_->setErrorCallback(std::bind(&TcpConnection::handleError, this));LOG_DEBUG << "TcpConnection::ctor[" << name_ << "] at " << this<< " fd=" << sockfd;socket_->setKeepAlive(true); }

設(shè)置了套接字可讀，可讀對應(yīng)的回調(diào)函數(shù)。當(dāng)client發(fā)送數(shù)據(jù)過來時，TcpConnection::handleRead被調(diào)用。

void TcpConnection::handleRead(Timestamp receiveTime) {loop_->assertInLoopThread();int savedErrno = 0;ssize_t n = inputBuffer_.readFd(channel_->fd(), &savedErrno);if (n > 0){//調(diào)用回調(diào)函數(shù),使用shared_from_this()得到自身的shared_ptr,延長了該對象的生命期,保證了它的生命期長過messageCallback_函數(shù),messageCallback_能安全的使用它messageCallback_(shared_from_this(), &inputBuffer_, receiveTime);}else if (n == 0){handleClose();}else{errno = savedErrno;LOG_SYSERR << "TcpConnection::handleRead";handleError();} }

? ?最終調(diào)用到了用戶設(shè)置的messageCallback_，到此流程大致分析完畢。

tcpserver: setNewConnectionCallback //新連接到達(新連接到達由acccptor處理)acceptor:setReadCallback //有可讀事件linsten:enableReading //注冊可讀事件echo:setConnectionCallback //連接到達setMessageCallback //數(shù)據(jù)到達TcpConnection：setReadCallback //可讀事件setWriteCallback //可寫事件

? ? 上面是各個類設(shè)置的對應(yīng)的回調(diào)函數(shù)，當(dāng)新連接達到時，acceptor處理，然后到tcpserver，然后到TcpConnection，在后面就是TcpConnection和echo進行交互。

總結(jié)

以上是生活随笔為你收集整理的muduo源码client/server通信流程的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。